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
python类ACCEPTED的实例源码
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)
