def test_send_samples_invalid_server_res(self, mock_cs):
mock_cs.side_effect = [None]
api = API.apiCalls.ApiCalls(
client_id="",
client_secret="",
base_URL="",
username="",
password=""
)
session_response = Foo()
setattr(session_response, "status_code", httplib.CONFLICT)
setattr(session_response, "text",
"An entity already exists with that identifier")
session_post = MagicMock(side_effect=[session_response])
session = Foo()
setattr(session, "post", session_post)
api.session = session
api.get_link = lambda x, y, targ_dict="": None
sample = API.apiCalls.Sample({"sampleProject": "1", "sampleName": "1"})
with self.assertRaises(API.apiCalls.SampleError) as err:
api.send_samples([sample])
self.assertTrue(str(session_response.status_code) + ": " +
session_response.text in str(err.exception))
python类CONFLICT的实例源码
def _makeConflictResponse(operation, retrievedlock, lock, oldlock, filename):
'''Generates and logs an HTTP 401 response in case of locks conflict'''
resp = flask.Response()
resp.headers['X-WOPI-Lock'] = retrievedlock if retrievedlock else ''
resp.status_code = httplib.CONFLICT
Wopi.log.info('msg="%s" filename="%s" token="%s", lock="%s" oldLock="%s" retrievedLock="%s" result="conflict"' % \
(operation.title(), filename, flask.request.args['access_token'][-20:], lock, oldlock, retrievedlock))
return resp
def check_resp_status_and_retry(resp, image_id, url):
# Note(Jesse): This branch sorts errors into those that are permanent,
# those that are ephemeral, and those that are unexpected.
if resp.status in (httplib.BAD_REQUEST, # 400
httplib.UNAUTHORIZED, # 401
httplib.PAYMENT_REQUIRED, # 402
httplib.FORBIDDEN, # 403
httplib.METHOD_NOT_ALLOWED, # 405
httplib.NOT_ACCEPTABLE, # 406
httplib.PROXY_AUTHENTICATION_REQUIRED, # 407
httplib.CONFLICT, # 409
httplib.GONE, # 410
httplib.LENGTH_REQUIRED, # 411
httplib.PRECONDITION_FAILED, # 412
httplib.REQUEST_ENTITY_TOO_LARGE, # 413
httplib.REQUEST_URI_TOO_LONG, # 414
httplib.UNSUPPORTED_MEDIA_TYPE, # 415
httplib.REQUESTED_RANGE_NOT_SATISFIABLE, # 416
httplib.EXPECTATION_FAILED, # 417
httplib.UNPROCESSABLE_ENTITY, # 422
httplib.LOCKED, # 423
httplib.FAILED_DEPENDENCY, # 424
httplib.UPGRADE_REQUIRED, # 426
httplib.NOT_IMPLEMENTED, # 501
httplib.HTTP_VERSION_NOT_SUPPORTED, # 505
httplib.NOT_EXTENDED, # 510
):
raise PluginError("Got Permanent Error response [%i] while "
"uploading image [%s] to glance [%s]"
% (resp.status, image_id, url))
# Nova service would process the exception
elif resp.status == httplib.NOT_FOUND: # 404
exc = XenAPI.Failure('ImageNotFound')
raise exc
# NOTE(nikhil): Only a sub-set of the 500 errors are retryable. We
# optimistically retry on 500 errors below.
elif resp.status in (httplib.REQUEST_TIMEOUT, # 408
httplib.INTERNAL_SERVER_ERROR, # 500
httplib.BAD_GATEWAY, # 502
httplib.SERVICE_UNAVAILABLE, # 503
httplib.GATEWAY_TIMEOUT, # 504
httplib.INSUFFICIENT_STORAGE, # 507
):
raise RetryableError("Got Ephemeral Error response [%i] while "
"uploading image [%s] to glance [%s]"
% (resp.status, image_id, url))
else:
# Note(Jesse): Assume unexpected errors are retryable. If you are
# seeing this error message, the error should probably be added
# to either the ephemeral or permanent error list.
raise RetryableError("Got Unexpected Error response [%i] while "
"uploading image [%s] to glance [%s]"
% (resp.status, image_id, url))
def rest_call(self, action, resource, data, headers, ignore_codes,
timeout=False):
good_first = sorted(self.servers, key=lambda x: x.failed)
first_response = None
for active_server in good_first:
ret = active_server.rest_call(action, resource, data, headers,
timeout,
reconnect=self.always_reconnect)
# If inconsistent, do a full synchronization
# if ret[0] == httplib.CONFLICT:
# if self.get_topo_function:
# data = self.get_topo_function(
# **self.get_topo_function_args)
# active_server.rest_call('PUT', TOPOLOGY_PATH, data,
# timeout=None)
# Store the first response as the error to be bubbled up to the
# user since it was a good server. Subsequent servers will most
# likely be cluster slaves and won't have a useful error for the
# user (e.g. 302 redirect to master)
if not first_response:
first_response = ret
if not self.server_failure(ret, ignore_codes):
active_server.failed = False
return ret
else:
try:
LOG.error(_LE('ServerProxy: %(action)s failure for '
'servers:%(server)r Response:'
'%(response)s'),
{'action': action,
'server': (active_server.server,
active_server.port),
'response': unicode(ret[3], "utf-8")})
LOG.error(_LE("ServerProxy: Error details: "
"status=%(status)d, reason=%(reason)r, "
"ret=%(ret)s, data=%(data)r"),
{'status': ret[0], 'reason': ret[1],
'ret': unicode(ret[2], "utf-8"),
'data': unicode(ret[3], "utf-8")})
except Exception as e:
LOG.error(_LE("fail to display info, err: %(e)s"),
{'e': e})
active_server.failed = True
# All servers failed, reset server list and try again next time
LOG.error(_('ServerProxy: %(action)s failure for all servers: '
'%(server)r'),
{'action': action,
'server': tuple((s.server,
s.port) for s in self.servers)})
return first_response
