python类CREATED的实例源码

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

        :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: A tuple with the server response and the deserialized created
                 object.

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

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

        return resp, self.deserialize(body)

def test_create_success_with_201_response_code(
        self, mock_client, mock_create):
        body = {
            "segment": {
                "name": "segment1",
                "service_type": "COMPUTE",
                "recovery_method": "auto",
                "description": "failover_segment for compute"
            }
        }
        fake_req = self.req
        fake_req.headers['Content-Type'] = 'application/json'
        fake_req.method = 'POST'
        fake_req.body = jsonutils.dump_as_bytes(body)
        resp = fake_req.get_response(self.app)
        self.assertEqual(http.CREATED, resp.status_code)

def test_create_success_with_201_response_code(
        self, mock_client, mock_create):
        body = {
            "host": {
                "name": "host-1", "type": "fake",
                "reserved": False,
                "on_maintenance": False,
                "control_attributes": "fake-control_attributes"
            }
        }
        fake_req = self.req
        fake_req.headers['Content-Type'] = 'application/json'
        fake_req.method = 'POST'
        fake_req.body = jsonutils.dump_as_bytes(body)
        resp = fake_req.get_response(self.app)
        self.assertEqual(http.CREATED, resp.status_code)

def test_accelerator_post(self, mock_uuid, mock_create):
        mock_uuid.return_value = self.ACCELERATOR_UUID

        body = gen_post_body(name='test_accelerator')
        headers = self.gen_headers(self.context)
        response = self.post_json('/accelerators', body, headers=headers)
        self.assertEqual(http_client.CREATED, response.status_int)
        response = response.json
        self.assertEqual(self.ACCELERATOR_UUID, response['uuid'])
        self.assertEqual(body['name'], response['name'])
        mock_create.assert_called_once_with(mock.ANY, mock.ANY, mock.ANY)

def _put(self, entity_id, body, **kwargs):
        url = self._build_path(entity_id)
        resp, body = super(Federation, self).put(url, body, **kwargs)
        self.expected_success(http_client.CREATED, resp.status)
        body = json.loads(body if six.PY2 else body.decode('utf-8'))
        return rest_client.ResponseBody(resp, body)

def test_assemble_node_failed(self, mock_request, mock_get_url,
                                  mock_delete_node):
        """Test allocate resource conflict when compose node"""
        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(
            fakes.fake_assemble_node_failed(), http_client.BAD_REQUEST)
        mock_request.side_effect = [fake_node_root_resp,
                                    fake_node_allocation_conflict,
                                    fake_node_detail,
                                    fake_node_assemble_failed]

        with self.assertRaises(exception.RedfishException):
            redfish.compose_node({"name": "test_node"})

        mock_delete_node.assert_called_once()

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 set_key(self, req, dpid, key, **_kwargs):
        def _set_val(dpid, key):
            try:
                val = req.json if req.body else {}
            except ValueError:
                return Response(status=http_client.BAD_REQUEST,
                                body='invalid syntax %s' % req.body)
            self.conf_switch.set_key(dpid, key, val)
            return None

        def _ret(_ret):
            return Response(status=http_client.CREATED)

        return self._do_key(dpid, key, _set_val, _ret)

def set_key(self, req, dpid, key, **_kwargs):
        def _set_val(dpid, key):
            try:
                val = req.json if req.body else {}
            except ValueError:
                return Response(status=http_client.BAD_REQUEST,
                                body='invalid syntax %s' % req.body)
            self.conf_switch.set_key(dpid, key, val)
            return None

        def _ret(_ret):
            return Response(status=http_client.CREATED)

        return self._do_key(dpid, key, _set_val, _ret)

def _request_unscoped_token(self):
        resp = self.saml2_client.send_service_provider_request(
            self.keystone_v3_endpoint, self.idp_id, self.protocol_id)
        self.assertEqual(http_client.OK, resp.status_code)
        saml2_authn_request = etree.XML(resp.content)

        relay_state = self._str_from_xml(
            saml2_authn_request, self.ECP_RELAY_STATE)
        sp_consumer_url = self._str_from_xml(
            saml2_authn_request, self.ECP_SERVICE_PROVIDER_CONSUMER_URL)

        # Perform the authn request to the identity provider
        resp = self.saml2_client.send_identity_provider_authn_request(
            saml2_authn_request, self.idp_url, self.username, self.password)
        self.assertEqual(http_client.OK, resp.status_code)
        saml2_idp_authn_response = etree.XML(resp.content)

        idp_consumer_url = self._str_from_xml(
            saml2_idp_authn_response, self.ECP_IDP_CONSUMER_URL)

        # Assert that both saml2_authn_request and saml2_idp_authn_response
        # have the same consumer URL.
        self.assertEqual(sp_consumer_url, idp_consumer_url)

        # Present the identity provider authn response to the service provider.
        resp = self.saml2_client.send_service_provider_saml2_authn_response(
            saml2_idp_authn_response, relay_state, idp_consumer_url)
        # Must receive a redirect from service provider to the URL where the
        # unscoped token can be retrieved.
        self.assertIn(resp.status_code,
                      [http_client.FOUND, http_client.SEE_OTHER])

        # We can receive multiple types of errors here, the response depends on
        # the mapping and the username used to authenticate in the Identity
        # Provider and also in the Identity Provider remote ID validation.
        # If everything works well, we receive an unscoped token.
        sp_url = resp.headers['location']
        resp = (
            self.saml2_client.send_service_provider_unscoped_token_request(
                sp_url))
        self.assertEqual(http_client.CREATED, resp.status_code)
        self.assertIn('X-Subject-Token', resp.headers)
        self.assertNotEmpty(resp.json())

        return resp

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)