Python falcon 模块，HTTP_204 实例源码

def on_get(self, req, resp):
        token = req.get_param('token', True)
        data = {}
        for key in self.data_keys:
            data[key] = req.get_param(key, True)

        if not self.validate_token(token, data):
            raise falcon.HTTPForbidden('Invalid token for these given values', '')

        endpoint = self.config['iris']['hook']['gmail_one_click']

        try:
            result = self.iclient.post(endpoint, data)
        except MaxRetryError:
            logger.exception('Hitting iris-api failed for gmail oneclick')
        else:
            if result.status == 204:
                resp.status = falcon.HTTP_204
                return
            else:
                logger.error('Unexpected status code from api %s for gmail oneclick', result.status)

        raise falcon.HTTPInternalServerError('Internal Server Error', 'Invalid response from API')

def test_cluster_restart_retrieve(self):
        """
        Verify retrieving a cluster restart.
        """
        with mock.patch('cherrypy.engine.publish') as _publish:
            manager = mock.MagicMock(StoreHandlerManager)
            _publish.return_value = [manager]

            test_cluster_restart = make_new(CLUSTER_RESTART)

            # Verify if the cluster restart exists the data is returned
            manager.get.return_value = test_cluster_restart
            body = self.simulate_request('/api/v0/cluster/development/restart')
            self.assertEqual(falcon.HTTP_200, self.srmock.status)
            self.assertEqual(json.loads(test_cluster_restart.to_json()), json.loads(body[0]))

            # Verify no cluster restart returns the proper result
            manager.get.side_effect = (
                test_cluster_restart,
                Exception)
            body = self.simulate_request('/api/v0/cluster/development/restart')
            self.assertEqual(falcon.HTTP_204, self.srmock.status)
            self.assertEqual([], body)  # Empty data

def test_cluster_upgrade_retrieve(self):
        """
        Verify retrieving a cluster upgrade.
        """
        with mock.patch('cherrypy.engine.publish') as _publish:
            manager = mock.MagicMock(StoreHandlerManager)
            _publish.return_value = [manager]

            test_cluster_upgrade = make_new(CLUSTER_UPGRADE)
            # Verify if the cluster upgrade exists the data is returned
            manager.get.return_value = test_cluster_upgrade
            body = self.simulate_request('/api/v0/cluster/development/upgrade')
            self.assertEqual(falcon.HTTP_200, self.srmock.status)
            self.assertEqual(json.loads(test_cluster_upgrade.to_json()), json.loads(body[0]))

            # Verify no cluster upgrade returns the proper result
            manager.reset_mock()
            manager.get.side_effect = (
                test_cluster_upgrade,
                Exception)

            body = self.simulate_request('/api/v0/cluster/development/upgrade')
            self.assertEqual(falcon.HTTP_204, self.srmock.status)
            self.assertEqual([], body)  # Empty data

def test_cluster_deploy_retrieve(self):
        """
        Verify retrieving a cluster deploy.
        """
        with mock.patch('cherrypy.engine.publish') as _publish:
            manager = mock.MagicMock(StoreHandlerManager)
            _publish.return_value = [manager]

            test_cluster_deploy = make_new(CLUSTER_DEPLOY)
            # Verify if the cluster deploy exists the data is returned
            manager.get.return_value = test_cluster_deploy
            body = self.simulate_request('/api/v0/cluster/development/deploy')
            self.assertEqual(falcon.HTTP_200, self.srmock.status)
            self.assertEqual(json.loads(test_cluster_deploy.to_json()), json.loads(body[0]))

            # Verify no cluster deploy returns the proper result
            manager.reset_mock()
            manager.get.side_effect = (
                test_cluster_deploy,
                Exception)

            body = self.simulate_request('/api/v0/cluster/development/deploy')
            self.assertEqual(falcon.HTTP_204, self.srmock.status)
            self.assertEqual([], body)  # Empty data

def test_health_endpoint(self):
        """
        Validate that /api/v1.0/health returns 204.
        """
        result = self.simulate_get('/api/v1.0/health')
        self.assertEqual(result.status, falcon.HTTP_204)

def on_get(self, req, resp, sid, id):
        self.called = True

        self.req, self.resp = req, resp
        resp.status = falcon.HTTP_204

    # Field names ordered the same as in uri template

def on_options(self, req, resp):
        # NOTE(kgriffs): The default responder returns 200
        resp.status = falcon.HTTP_204

        # NOTE(kgriffs): This is incorrect, but only return GET so
        # that we can verify that the default OPTIONS responder has
        # been overridden.
        resp.set_header('allow', 'GET')

def test_get(self, client, resource_things):
        client.app.add_route('/things', resource_things)
        client.app.add_route('/things/{id}/stuff/{sid}', resource_things)
        response = client.simulate_request(path='/things/42/stuff/57')
        assert response.status == falcon.HTTP_204
        assert resource_things.called

def test_on_options(self, client):
        response = client.simulate_request(path='/misc', method='OPTIONS')
        assert response.status == falcon.HTTP_204

        headers = response.headers
        assert headers['allow'] == 'GET'

def on_get(self, req, resp, name):
        """
        Handles GET (or "status") requests for a tree image deployment
        across 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 undergoing deployment.
        :type name: str
        """
        if not util.etcd_cluster_exists(name):
            self.logger.info(
                'Deploy GET requested for nonexistent cluster {0}'.format(
                    name))
            resp.status = falcon.HTTP_404
            return

        try:
            store_manager = cherrypy.engine.publish('get-store-manager')[0]
            cluster_deploy = store_manager.get(ClusterDeploy.new(name=name))
            self.logger.debug('Found ClusterDeploy for {0}'.format(name))
        except:
            # Return "204 No Content" if we have no status,
            # meaning no deployment is in progress.  The client
            # can't be expected to know that, so it's not a
            # client error (4xx).
            self.logger.debug((
                'Deploy GET requested for {0} but no deployment '
                'has ever been executed.').format(name))

            resp.status = falcon.HTTP_204
            return

        resp.status = falcon.HTTP_200
        req.context['model'] = cluster_deploy

def on_get(self, req, resp, name):
        """
        Handles GET (or "status") requests for a Cluster restart.

        :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 restarted.
        :type name: str
        """
        if not util.etcd_cluster_exists(name):
            self.logger.info(
                'Restart GET requested for nonexistent cluster {0}'.format(
                    name))
            resp.status = falcon.HTTP_404
            return

        try:
            store_manager = cherrypy.engine.publish('get-store-manager')[0]
            cluster_restart = store_manager.get(ClusterRestart.new(name=name))
        except:
            # Return "204 No Content" if we have no status,
            # meaning no restart is in progress.  The client
            # can't be expected to know that, so it's not a
            # client error (4xx).
            self.logger.debug((
                'Restart GET requested for {0} but no restart '
                'has ever been executed.').format(name))
            resp.status = falcon.HTTP_204
            return
        resp.status = falcon.HTTP_200
        req.context['model'] = cluster_restart

def on_options(self, req, resp):
        "Handles OPTIONS requests."
        resp.status = falcon.HTTP_204
        pass