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

def add(self):
        params = None
        try:
            params = utils.CNIParameters(flask.request.get_json())
            LOG.debug('Received addNetwork request. CNI Params: %s', params)
            vif = self.plugin.add(params)
            data = jsonutils.dumps(vif.obj_to_primitive())
        except exceptions.ResourceNotReady as e:
            LOG.error("Timed out waiting for requested pod to appear in "
                      "registry: %s.", e)
            return '', httplib.GATEWAY_TIMEOUT, self.headers
        except Exception:
            LOG.exception('Error when processing addNetwork request. CNI '
                          'Params: %s', params)
            return '', httplib.INTERNAL_SERVER_ERROR, self.headers
        return data, httplib.ACCEPTED, self.headers

def test_create_success_with_201_response_code(
        self, mock_client, mock_create):
        body = {
            "notification": {
                "hostname": "fake_host",
                "payload": {
                    "event": "STOPPED",
                    "host_status": "NORMAL",
                    "cluster_status": "ONLINE"
                },
                "type": "VM",
                "generated_time": NOW
            }
        }
        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.ACCEPTED, resp.status_code)

def test_resource_headers_are_utf8(self):
        resp = webob.Response(status_int=http.ACCEPTED)
        resp.headers['x-header1'] = 1
        resp.headers['x-header2'] = u'header2'
        resp.headers['x-header3'] = u'header3'

        class Controller(object):
            def index(self, req):
                return resp

        req = webob.Request.blank('/tests')
        app = fakes.TestRouter(Controller())
        response = req.get_response(app)
        for val in response.headers.values():
            # All headers must be utf8
            self.assertThat(val, matchers.EncodedByUTF8())
        self.assertEqual(b'1', response.headers['x-header1'])
        self.assertEqual(b'header2', response.headers['x-header2'])
        self.assertEqual(b'header3', response.headers['x-header3'])

def _add(self, params):
        resp = self._make_request('addNetwork', params, httplib.ACCEPTED)
        vif = base.VersionedObject.obj_from_primitive(resp.json())
        return self._vif_data(vif)

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_console_mode(self, node_uuid, enabled):
        """Start and stop the node console.

        :param node_uuid: Unique identifier of the node in UUID format.
        :param enabled: Boolean value; whether to enable or disable the
                        console.

        """

        enabled = {'enabled': enabled}
        resp, body = self._put_request('nodes/%s/states/console' % node_uuid,
                                       enabled)
        self.expected_success(http_client.ACCEPTED, resp.status)
        return resp, body

def delete_switch(self, _req, dpid, **_kwargs):
        def _delete_switch(dpid):
            self.conf_switch.del_dpid(dpid)
            return None

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

        return self._do_switch(dpid, _delete_switch, _ret)

def test_modified_code(self):
        robj = wsgi.ResponseObject({})
        robj._default_code = http.ACCEPTED
        self.assertEqual(robj.code, http.ACCEPTED)

def test_override_modified_code(self):
        robj = wsgi.ResponseObject({}, code=http.NOT_FOUND)
        robj._default_code = http.ACCEPTED
        self.assertEqual(robj.code, http.NOT_FOUND)

def delete_switch(self, _req, dpid, **_kwargs):
        def _delete_switch(dpid):
            self.conf_switch.del_dpid(dpid)
            return None

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

        return self._do_switch(dpid, _delete_switch, _ret)