Python httplib 模块，CONFLICT 实例源码

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))

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