Python six.moves.http_client 模块，NO_CONTENT 实例源码

def delete(self):
        params = None
        try:
            params = utils.CNIParameters(flask.request.get_json())
            LOG.debug('Received delNetwork request. CNI Params: %s', params)
            self.plugin.delete(params)
        except exceptions.ResourceNotReady as e:
            # NOTE(dulek): It's better to ignore this error - most of the time
            #              it will happen when pod is long gone and kubelet
            #              overzealously tries to delete it from the network.
            #              We cannot really do anything without VIF annotation,
            #              so let's just tell kubelet to move along.
            LOG.warning("Timed out waiting for requested pod to appear in "
                        "registry: %s. Ignoring.", e)
            return '', httplib.NO_CONTENT, self.headers
        except Exception:
            LOG.exception('Error when processing delNetwork request. CNI '
                          'Params: %s.', params)
            return '', httplib.INTERNAL_SERVER_ERROR, self.headers
        return '', httplib.NO_CONTENT, self.headers

def _create_request_no_response_body(self, resource, object_dict):
        """Create an object of the specified type.

           Do not expect any body in the response.

        :param resource: The name of the REST resource, e.g., 'nodes'.
        :param object_dict: A Python dict that represents an object of the
                            specified type.
        :returns: The server response.
        """

        body = self.serialize(object_dict)
        uri = self._get_uri(resource)

        resp, body = self.post(uri, body=body)
        self.expected_success(http_client.NO_CONTENT, resp.status)

        return resp

def test_reset_node_success(self, mock_get_url, mock_request):
        """Test successfully reset node status"""
        mock_get_url.return_value = '/redfish/v1/Nodes'
        fake_node_detail = fakes.mock_request_get(
            fakes.fake_node_detail(), http_client.OK)
        fake_node_action_resp = fakes.mock_request_get(
            {}, http_client.NO_CONTENT)
        mock_request.side_effect = [fake_node_detail, fake_node_action_resp]

        result = redfish.reset_node("1", {"Reset": {"Type": "On"}})
        expected = exception.confirmation(
            confirm_code="Reset Composed Node",
            confirm_detail="This composed node has been set to 'On' "
                           "successfully.")

        self.assertEqual(expected, result)

def test_set_boot_source_success(self, mock_get_url, mock_request):
        """Test successfully reset node status"""
        mock_get_url.return_value = '/redfish/v1/Nodes'
        fake_node_detail = fakes.mock_request_get(
            fakes.fake_node_detail(), http_client.OK)
        fake_node_action_resp = fakes.mock_request_get(
            {}, http_client.NO_CONTENT)
        mock_request.side_effect = [fake_node_detail, fake_node_action_resp]

        result = redfish.set_boot_source(
            "1", {"Boot": {"Enabled": "Once", "Target": "Hdd"}})
        expected = exception.confirmation(
            confirm_code="Set Boot Source of Composed Node",
            confirm_detail="The boot source of composed node has been set to "
                           "'{0}' with enabled state '{1}' successfully."
                           .format("Hdd", "Once"))

        self.assertEqual(expected, result)

def json_request(self, method, conn_url, **kwargs):
        kwargs.setdefault('headers', {})
        kwargs['headers'].setdefault('Content-Type', 'application/json')
        kwargs['headers'].setdefault('Accept', 'application/json')
        if 'body' in kwargs:
            kwargs['body'] = jsonutils.dump_as_bytes(kwargs['body'])

        resp, body_iter = self._http_request(conn_url, method, **kwargs)
        content_type = resp.headers.get('Content-Type')
        if(resp.status_code in (http_client.NO_CONTENT,
                                http_client.RESET_CONTENT)
            or content_type is None):
            return resp, list()

        if 'application/json' in content_type:
            body = ''.join([chunk for chunk in body_iter])
            try:
                body = jsonutils.loads(body)
            except ValueError:
                LOG.error(_LE('Could not decode response body as JSON'))
        else:
            body = None
        return resp, body

def json_request(self, method, url, **kwargs):
        kwargs.setdefault('headers', {})
        kwargs['headers'].setdefault('Content-Type', 'application/json')
        kwargs['headers'].setdefault('Accept', 'application/json')

        if 'body' in kwargs:
            kwargs['data'] = jsonutils.dump_as_bytes(kwargs.pop('body'))

        resp = self._http_request(url, method, **kwargs)
        body = resp.content
        content_type = resp.headers.get('content-type', None)
        status = resp.status_code
        if (status in (
                http_client.NO_CONTENT, http_client.RESET_CONTENT) or
                content_type is None):
            return resp, list()
        if 'application/json' in content_type:
            try:
                body = resp.json()
            except ValueError:
                LOG.error(_LE('Could not decode response body as JSON'))
        else:
            body = None

        return resp, body

def _delete(self, params):
        self._make_request('delNetwork', params, httplib.NO_CONTENT)

def _delete(self, entity_id, **kwargs):
        url = self._build_path(entity_id)
        resp, body = super(Federation, self).delete(url, **kwargs)
        self.expected_success(http_client.NO_CONTENT, resp.status)
        return rest_client.ResponseBody(resp, body)

def _delete_request(self, resource, uuid):
        """Delete specified object.

        :param resource: The name of the REST resource, e.g., 'nodes'.
        :param uuid: The unique identifier of an object in UUID format.
        :returns: A tuple with the server response and the response body.

        """
        uri = self._get_uri(resource, uuid)

        resp, body = self.delete(uri)
        self.expected_success(http_client.NO_CONTENT, resp.status)
        return resp, body

def _put_request(self, resource, put_object):
        """Update specified object with JSON-patch."""
        uri = self._get_uri(resource)
        put_body = json.dumps(put_object)

        resp, body = self.put(uri, body=put_body)
        self.expected_success([http_client.ACCEPTED, http_client.NO_CONTENT],
                              resp.status)
        return resp, body

def set_node_boot_device(self, node_uuid, boot_device, persistent=False):
        """Set the boot device of the specified node.

        :param node_uuid: The unique identifier of the node.
        :param boot_device: The boot device name.
        :param persistent: Boolean value. True if the boot device will
                           persist to all future boots, False if not.
                           Default: False.

        """
        request = {'boot_device': boot_device, 'persistent': persistent}
        resp, body = self._put_request('nodes/%s/management/boot_device' %
                                       node_uuid, request)
        self.expected_success(http_client.NO_CONTENT, resp.status)
        return body

def vif_detach(self, node_uuid, vif_id):
        """Detach a VIF from a node

        :param node_uuid: Unique identifier of the node in UUID format.
        :param vif_id: An ID representing the VIF
        """
        resp, body = self._delete_request('nodes/%s/vifs' % node_uuid, vif_id)
        self.expected_success(http_client.NO_CONTENT, resp.status)
        return resp, body

def test_delete_composednode_ok(self, mock_request, mock_get_url):
        mock_get_url.return_value = '/redfish/v1/Nodes'
        delete_result = fakes.fake_delete_composednode_ok()
        fake_delete_response = fakes.mock_request_get(delete_result,
                                                      http_client.NO_CONTENT)
        mock_request.return_value = fake_delete_response
        result = redfish.delete_composed_node(101)
        mock_request.assert_called_with('/redfish/v1/Nodes/101', 'DELETE')
        expected = {
            "code": "DELETED",
            "detail": "This composed node has been deleted successfully.",
            "request_id": exception.FAKE_REQUEST_ID,
        }

        self.assertEqual(expected, result)

def test_assemble_node_success(self, mock_request, mock_get_url,
                                   mock_delete_node, mock_get_node_by_id):
        """Test compose node successfully"""
        CONF.set_override('url', 'http://localhost:8442/', group='podm')
        mock_get_url.return_value = '/redfish/v1/Nodes'

        # Fake response for getting nodes root
        fake_node_root_resp = fakes.mock_request_get(fakes.fake_nodes_root(),
                                                     http_client.OK)
        # Fake response for allocating node
        fake_node_allocation_conflict = mock.MagicMock()
        fake_node_allocation_conflict.status_code = http_client.CREATED
        fake_node_allocation_conflict.headers['Location'] = \
            os.path.normpath("/".join([CONF.podm.url, 'redfish/v1/Nodes/1']))

        # Fake response for getting url of node assembling
        fake_node_detail = fakes.mock_request_get(fakes.fake_node_detail(),
                                                  http_client.OK)

        # Fake response for assembling node
        fake_node_assemble_failed = fakes.mock_request_get(
            {}, http_client.NO_CONTENT)
        mock_request.side_effect = [fake_node_root_resp,
                                    fake_node_allocation_conflict,
                                    fake_node_detail,
                                    fake_node_assemble_failed]

        redfish.compose_node({"name": "test_node"})

        mock_delete_node.assert_not_called()
        mock_get_node_by_id.assert_called_once()

def test_node_action_request(self, mock_action, m_node, mock_connection):
        req = {
            "Reset": {
                "Type": "On"
            }
        }
        mock_action.return_value = None
        resp = self.app.post('/v1/nodes/fake-node/action',
                             content_type='application/json',
                             data=json.dumps(req))
        mock_action.assert_called_once_with('fake-node', req)
        self.assertEqual(http_client.NO_CONTENT, resp.status_code)

def test_node_action_detach_request(self, mock_node, mock_action,
                                        mock_connection):
        req = {
            "detach": {
                "resource_id": "test-device-1"
            }
        }
        mock_action.return_value = None
        resp = self.app.post('/v1/nodes/fake-node/action',
                             content_type='application/json',
                             data=json.dumps(req))
        mock_action.assert_called_once_with('fake-node', req)
        self.assertEqual(http_client.NO_CONTENT, resp.status_code)

def post(self, node_uuid):
        return utils.make_response(
            http_client.NO_CONTENT,
            nodes.Node(node_id=node_uuid).node_action(node_uuid,
                                                      request.get_json()))

def delete_composed_node(nodeid):
    nodes_url = get_base_resource_url("Nodes")
    delete_url = nodes_url + '/' + str(nodeid)
    resp = send_request(delete_url, "DELETE")
    if resp.status_code == http_client.NO_CONTENT:
        # we should return 200 status code instead of 204, because 204 means
        # 'No Content', the message in resp_dict will be ignored in that way
        return exception.confirmation(
            confirm_code="DELETED",
            confirm_detail="This composed node has been deleted successfully.")
    else:
        raise exception.RedfishException(resp.json(),
                                         status_code=resp.status_code)

def test_validate_href(self, head_mock):
        href = 'http://1.2.3.4/abc.iso'
        response = head_mock.return_value
        response.status_code = http_client.OK
        utils.validate_href(href)
        head_mock.assert_called_once_with(href)
        response.status_code = http_client.NO_CONTENT
        self.assertRaises(exception.ImageRefValidationFailed,
                          utils.validate_href,
                          href)
        response.status_code = http_client.BAD_REQUEST
        self.assertRaises(exception.ImageRefValidationFailed,
                          utils.validate_href, href)

def json_request(self, method, url, **kwargs):
        kwargs.setdefault('headers', {})
        kwargs['headers'].setdefault('Content-Type', 'application/json')
        kwargs['headers'].setdefault('Accept', 'application/json')

        if 'body' in kwargs:
            kwargs['body'] = jsonutils.dump_as_bytes(kwargs['body'])

        resp, body_iter = self._http_request(url, method, **kwargs)
        content_type = resp.headers.get('Content-Type')

        if (resp.status_code in (
                http_client.NO_CONTENT,
                http_client.RESET_CONTENT) or content_type is None):
            return resp, list()

        if 'application/json' in content_type:
            body = ''.join([chunk for chunk in body_iter])
            try:
                body = jsonutils.loads(body)
            except ValueError:
                LOG.error(_LE('Could not decode response body as JSON'))
        else:
            body = None

        return resp, body

def test_delete_segment_with_204_status(self, mock_client, mock_delete):
        url = '/v1/segments/%s' % uuidsentinel.fake_segment
        fake_req = fakes.HTTPRequest.blank(url, use_admin_context=True)
        fake_req.headers['Content-Type'] = 'application/json'
        fake_req.method = 'DELETE'
        resp = fake_req.get_response(self.app)
        self.assertEqual(http.NO_CONTENT, resp.status_code)

def test_delete_host_with_204_status(self, mock_client, mock_delete):
        url = '/v1/segments/%(segment)s/hosts/%(host)s' % {
            'segment': uuidsentinel.fake_segment1,
            'host': uuidsentinel.fake_host_1
        }
        fake_req = fakes.HTTPRequest.blank(url, use_admin_context=True)
        fake_req.headers['Content-Type'] = 'application/json'
        fake_req.method = 'DELETE'
        resp = fake_req.get_response(self.app)
        self.assertEqual(http.NO_CONTENT, resp.status_code)

def compose_node(request_body):
    """Compose new node through podm api.

    :param request_body: The request content to compose new node, which should
                         follow podm format. Valence api directly pass it to
                         podm right now.
    :returns: The numeric index of new composed node.
    """

    # Get url of allocating resource to node
    nodes_url = get_base_resource_url('Nodes')
    resp = send_request(nodes_url, 'GET')
    if resp.status_code != http_client.OK:
        LOG.error('Unable to query ' + nodes_url)
        raise exception.RedfishException(resp.json(),
                                         status_code=resp.status_code)
    respdata = resp.json()
    allocate_url = respdata['Actions']['#ComposedNodeCollection.Allocate'][
        'target']

    # Allocate resource to this node
    LOG.debug('Allocating Node: {0}'.format(request_body))
    allocate_resp = send_request(allocate_url, 'POST',
                                 headers={'Content-type': 'application/json'},
                                 json=request_body)
    if allocate_resp.status_code != http_client.CREATED:
        # Raise exception if allocation failed
        raise exception.RedfishException(allocate_resp.json(),
                                         status_code=allocate_resp.status_code)

    # Allocated node successfully
    # node_url -- relative redfish url e.g redfish/v1/Nodes/1
    node_url = allocate_resp.headers['Location'].lstrip(CONF.podm.url)
    # node_index -- numeric index of new node e.g 1
    node_index = node_url.split('/')[-1]
    LOG.debug('Successfully allocated node:' + node_url)

    # Get url of assembling node
    resp = send_request(node_url, "GET")
    respdata = resp.json()
    assemble_url = respdata['Actions']['#ComposedNode.Assemble']['target']

    # Assemble node
    LOG.debug('Assembling Node: {0}'.format(assemble_url))
    assemble_resp = send_request(assemble_url, "POST")

    if assemble_resp.status_code != http_client.NO_CONTENT:
        # Delete node if assemble failed
        delete_composed_node(node_index)
        raise exception.RedfishException(assemble_resp.json(),
                                         status_code=resp.status_code)
    else:
        # Assemble successfully
        LOG.debug('Successfully assembled node: ' + node_url)

    # Return new composed node index
    return get_node_by_id(node_index)