我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用httplib2.Response()。
def request(self, uri, method='GET', body=None, headers=None, redirections=1, connection_type=None): resp, content = self._iterable.pop(0) if content == 'echo_request_headers': content = headers elif content == 'echo_request_headers_as_json': content = json.dumps(headers) elif content == 'echo_request_body': if hasattr(body, 'read'): content = body.read() else: content = body elif content == 'echo_request_uri': content = uri if isinstance(content, six.text_type): content = content.encode('utf-8') return httplib2.Response(resp), content
def request(self, uri, method='GET', body=None, headers=None, redirections=1, connection_type=None): resp, content = self._iterable.pop(0) if content == 'echo_request_headers': content = headers elif content == 'echo_request_headers_as_json': content = simplejson.dumps(headers) elif content == 'echo_request_body': if hasattr(body, 'read'): content = body.read() else: content = body elif content == 'echo_request_uri': content = uri return httplib2.Response(resp), content
def __init__(self, resp, content, postproc): """Constructor for HttpRequestMock Args: resp: httplib2.Response, the response to emulate coming from the request content: string, the response body postproc: callable, the post processing function usually supplied by the model class. See model.JsonModel.response() as an example. """ self.resp = resp self.content = content self.postproc = postproc if resp is None: self.resp = httplib2.Response({'status': 200, 'reason': 'OK'}) if 'reason' in self.resp: self.resp.reason = self.resp['reason']
def request(self, uri, method="GET", body=None, headers=None, redirections=5): path = urlparse.urlparse(uri)[2] fname = os.path.join(HTTP_SRC_DIR, path[1:]) if not os.path.exists(fname): index = self.hit_counter.get(fname, 1) if os.path.exists(fname + "." + str(index)): self.hit_counter[fname] = index + 1 fname = fname + "." + str(index) if os.path.exists(fname): f = file(fname, "r") response = message_from_file(f) f.close() body = response.get_payload() response_headers = httplib2.Response(response) return (response_headers, body) else: return (httplib2.Response({"status": "404"}), "")
def test_delete_sg_rule_with_error(self, mock_delete, mock_wait): http_error = gce_errors.HttpError( httplib2.Response({ 'status': 404, 'reason': 'Not Found' }), content='') mock_delete.side_effect = http_error mock_wait.side_effect = gce_mock.wait_for_operation sg_rule = self.get_fake_sg_rule() self.assertIsNone( self._driver._delete_secgrp_rule(self.context, sg_rule['id'])) mock_delete.assert_called_once_with(self._driver.gce_svc, self._driver.gce_project, "secgrp-" + sg_rule['id']) mock_wait.assert_not_called()
def mock_valid_requests(self, uri, *args, **kwargs): url = urlparse(uri) # https://dev.twitter.com/oauth/reference/post/oauth/request_token if url[1] == 'api.twitter.com' and url[2] == '/oauth/request_token': res = httplib2.Response(dict(status = 200)) content = b'oauth_token=foo&oauth_token_secret=bar&oauth_callback_confirmed=true' # https://dev.twitter.com/oauth/reference/post/oauth/access_token if url[1] == 'api.twitter.com' and url[2] == '/oauth/access_token': res = httplib2.Response(dict(status = 200)) content = b'oauth_token=foo&oauth_token_secret=bar&user_id=987&screen_name=test' # https://developers.facebook.com/docs/facebook-login/manually-build-a-login-flow#confirm if url[1] == 'graph.facebook.com' and url[2].endswith('/oauth/access_token'): res = httplib2.Response(dict(status = 200)) content = b'{"access_token":"foobar","token_type":"bearer","expires_in":5117097}' # https://developers.facebook.com/docs/graph-api/using-graph-api if url[1] == 'graph.facebook.com' and url[2] == '/me': res = httplib2.Response(dict(status = 200)) content = b'{"id":"987","name":"test"}' return (res, content)
def testHttpError(self): http_error_code = 403 with patch('airflow.contrib.operators.mlengine_operator.MLEngineHook') \ as mock_hook: hook_instance = mock_hook.return_value hook_instance.create_job.side_effect = errors.HttpError( resp=httplib2.Response({ 'status': http_error_code }), content=b'Forbidden') with self.assertRaises(errors.HttpError) as context: training_op = MLEngineTrainingOperator( **self.TRAINING_DEFAULT_ARGS) training_op.execute(None) mock_hook.assert_called_with( gcp_conn_id='google_cloud_default', delegate_to=None) # Make sure only 'create_job' is invoked on hook instance self.assertEquals(len(hook_instance.mock_calls), 1) hook_instance.create_job.assert_called_with( 'test-project', self.TRAINING_INPUT, ANY) self.assertEquals(http_error_code, context.exception.resp.status)
def testDigestObjectStale(self): credentials = ('joe', 'password') host = None request_uri = '/projects/httplib2/test/digest/' headers = {} response = httplib2.Response({}) response['www-authenticate'] = ( 'Digest realm="myrealm", ' 'nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", ' 'algorithm=MD5, qop="auth", stale=true') response.status = 401 content = b"" d = httplib2.DigestAuthentication( credentials, host, request_uri, headers, response, content, None) # Returns true to force a retry self.assertTrue(d.response(response, content))
def testDigestObjectAuthInfo(self): credentials = ('joe', 'password') host = None request_uri = '/projects/httplib2/test/digest/' headers = {} response = httplib2.Response({}) response['www-authenticate'] = ( 'Digest realm="myrealm", ' 'nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", ' 'algorithm=MD5, qop="auth", stale=true') response['authentication-info'] = 'nextnonce="fred"' content = b"" d = httplib2.DigestAuthentication( credentials, host, request_uri, headers, response, content, None) # Returns true to force a retry self.assertFalse(d.response(response, content)) self.assertEqual('fred', d.challenge['nonce']) self.assertEqual(1, d.challenge['nc'])
def _map_response(response, decode=False): """Maps a urllib3 response to a httplib/httplib2 Response.""" # This causes weird deepcopy errors, so it's commented out for now. # item._urllib3_response = response item = httplib2.Response(response.getheaders()) item.status = response.status item['status'] = str(item.status) item.reason = response.reason item.version = response.version # httplib2 expects the content-encoding header to be stripped and the # content length to be the length of the uncompressed content. # This does not occur for 'HEAD' requests. if decode and item.get('content-encoding') in ['gzip', 'deflate']: item['content-length'] = str(len(response.data)) item['-content-encoding'] = item.pop('content-encoding') return item
def test_put_upload_sends_content_type_and_length_headers(self): response = httplib2.Response({'status': 200}) mock_request = self.patch(boot_resources_create, 'http_request') mock_request.return_value = (response, b'') action = self.make_boot_resources_create_action() self.patch(action, 'sign') data = factory.make_bytes() action.put_upload(sentinel.upload_uri, data) headers = { 'Content-Type': 'application/octet-stream', 'Content-Length': '%s' % len(data), } self.assertThat( mock_request, MockCalledOnceWith( sentinel.upload_uri, 'PUT', body=ANY, headers=headers, insecure=False))
def test__prints_textual_response_with_success_msg(self): # When the response has a status code of 2XX, and the response body is # textual print_response_content() will print a success message to the # TTY. status_code = random.randrange(200, 300) response = httplib2.Response({ 'status': status_code, 'content': b"Lorem ipsum dolor sit amet.", 'content-type': 'text/unicode', }) buf = io.BytesIO() self.patch(buf, 'isatty').return_value = True utils.print_response_content(response, response['content'], buf) self.assertEqual( b"Success.\n" b"Machine-readable output follows:\n" + response['content'] + b"\n", buf.getvalue())
def test_fetch_api_description_not_okay(self): # If the response is not 200 OK, fetch_api_description throws toys. content = factory.make_name("content") request = self.patch(httplib2.Http, "request") response = httplib2.Response({}) response.status = http.client.BAD_REQUEST response.reason = http.client.responses[http.client.BAD_REQUEST] request.return_value = response, json.dumps(content) error = self.assertRaises( CommandError, api.fetch_api_description, "http://example.com/api/2.0/") error_expected = "%d %s:\n%s" % ( http.client.BAD_REQUEST, http.client.responses[http.client.BAD_REQUEST], json.dumps(content)) self.assertEqual(error_expected, "%s" % error)
def request(self, uri, method="GET", body=None, headers=None, redirections=5, connection_type=None): if not self.return_type: fake_headers = httplib2.Response(headers) return_obj = { 'uri': uri, 'method': method, 'body': body, 'headers': headers } return (fake_headers, return_obj) # return (headers, return_obj) elif isinstance(self.return_type, int): body = "fake_body" header_info = { 'content-type': 'text/plain', 'status': str(self.return_type), 'content-length': len(body) } resp_header = httplib2.Response(header_info) return (resp_header, body) else: msg = "unsupported return type %s" % self.return_type raise TypeError(msg)
def _retry_request(http, num_retries, req_type, sleep, rand, uri, method, *args, **kwargs): """Retries an HTTP request multiple times while handling errors. If after all retries the request still fails, last error is either returned as return value (for HTTP 5xx errors) or thrown (for ssl.SSLError). Args: http: Http object to be used to execute request. num_retries: Maximum number of retries. req_type: Type of the request (used for logging retries). sleep, rand: Functions to sleep for random time between retries. uri: URI to be requested. method: HTTP method to be used. args, kwargs: Additional arguments passed to http.request. Returns: resp, content - Response from the http request (may be HTTP 5xx). """ resp = None for retry_num in range(num_retries + 1): if retry_num > 0: sleep(rand() * 2**retry_num) logging.warning( 'Retry #%d for %s: %s %s%s' % (retry_num, req_type, method, uri, ', following status: %d' % resp.status if resp else '')) try: resp, content = http.request(uri, method, *args, **kwargs) except ssl.SSLError: if retry_num == num_retries: raise else: continue if resp.status < 500: break return resp, content
def add_response_callback(self, cb): """add_response_headers_callback Args: cb: Callback to be called on receiving the response headers, of signature: def cb(resp): # Where resp is an instance of httplib2.Response """ self.response_callbacks.append(cb)
def _process_response(self, resp, content): """Process the response from a single chunk upload. Args: resp: httplib2.Response, the response object. content: string, the content of the response. Returns: (status, body): (ResumableMediaStatus, object) The body will be None until the resumable media is fully uploaded. Raises: googleapiclient.errors.HttpError if the response was not a 2xx or a 308. """ if resp.status in [200, 201]: self._in_error_state = False return None, self.postproc(resp, content) elif resp.status == 308: self._in_error_state = False # A "308 Resume Incomplete" indicates we are not done. self.resumable_progress = int(resp['range'].split('-')[1]) + 1 if 'location' in resp: self.resumable_uri = resp['location'] else: self._in_error_state = True raise HttpError(resp, content, uri=self.uri) return (MediaUploadProgress(self.resumable_progress, self.resumable.size()), None)
def __init__(self, callback=None, batch_uri=None): """Constructor for a BatchHttpRequest. Args: callback: callable, A callback to be called for each response, of the form callback(id, response, exception). The first parameter is the request id, and the second is the deserialized response object. The third is an googleapiclient.errors.HttpError exception object if an HTTP error occurred while processing the request, or None if no error occurred. batch_uri: string, URI to send batch requests to. """ if batch_uri is None: batch_uri = 'https://www.googleapis.com/batch' self._batch_uri = batch_uri # Global callback to be called for each individual response in the batch. self._callback = callback # A map from id to request. self._requests = {} # A map from id to callback. self._callbacks = {} # List of request ids, in the order in which they were added. self._order = [] # The last auto generated id. self._last_auto_id = 0 # Unique ID on which to base the Content-ID headers. self._base_id = None # A map from request id to (httplib2.Response, content) response pairs self._responses = {} # A map of id(Credentials) that have been refreshed. self._refreshed_credentials = {}
def _deserialize_response(self, payload): """Convert string into httplib2 response and content. Args: payload: string, headers and body as a string. Returns: A pair (resp, content), such as would be returned from httplib2.request. """ # Strip off the status line status_line, payload = payload.split('\n', 1) protocol, status, reason = status_line.split(' ', 2) # Parse the rest of the response parser = FeedParser() parser.feed(payload) msg = parser.close() msg['status'] = status # Create httplib2.Response from the parsed headers. resp = httplib2.Response(msg) resp.reason = reason resp.version = int(protocol.split('/', 1)[1].replace('.', '')) content = payload.split('\r\n\r\n', 1)[1] return resp, content
def __init__(self, responses, check_unexpected=False): """Constructor for RequestMockBuilder The constructed object should be a callable object that can replace the class HttpResponse. responses - A dictionary that maps methodIds into tuples of (httplib2.Response, content). The methodId comes from the 'rpcName' field in the discovery document. check_unexpected - A boolean setting whether or not UnexpectedMethodError should be raised on unsupplied method. """ self.responses = responses self.check_unexpected = check_unexpected
def request(self, uri, method='GET', body=None, headers=None, redirections=1, connection_type=None): self.uri = uri self.method = method self.body = body self.headers = headers return httplib2.Response(self.response_headers), self.data
def _process_response(self, resp, content): """Process the response from a single chunk upload. Args: resp: httplib2.Response, the response object. content: string, the content of the response. Returns: (status, body): (ResumableMediaStatus, object) The body will be None until the resumable media is fully uploaded. Raises: apiclient.errors.HttpError if the response was not a 2xx or a 308. """ if resp.status in [200, 201]: self._in_error_state = False return None, self.postproc(resp, content) elif resp.status == 308: self._in_error_state = False # A "308 Resume Incomplete" indicates we are not done. self.resumable_progress = int(resp['range'].split('-')[1]) + 1 if 'location' in resp: self.resumable_uri = resp['location'] else: self._in_error_state = True raise HttpError(resp, content, uri=self.uri) return (MediaUploadProgress(self.resumable_progress, self.resumable.size()), None)
def __init__(self, callback=None, batch_uri=None): """Constructor for a BatchHttpRequest. Args: callback: callable, A callback to be called for each response, of the form callback(id, response, exception). The first parameter is the request id, and the second is the deserialized response object. The third is an apiclient.errors.HttpError exception object if an HTTP error occurred while processing the request, or None if no error occurred. batch_uri: string, URI to send batch requests to. """ if batch_uri is None: batch_uri = 'https://www.googleapis.com/batch' self._batch_uri = batch_uri # Global callback to be called for each individual response in the batch. self._callback = callback # A map from id to request. self._requests = {} # A map from id to callback. self._callbacks = {} # List of request ids, in the order in which they were added. self._order = [] # The last auto generated id. self._last_auto_id = 0 # Unique ID on which to base the Content-ID headers. self._base_id = None # A map from request id to (httplib2.Response, content) response pairs self._responses = {} # A map of id(Credentials) that have been refreshed. self._refreshed_credentials = {}
def request(http, uri, method='GET', body=None, headers=None, redirections=httplib2.DEFAULT_MAX_REDIRECTS, connection_type=None): """Make an HTTP request with an HTTP object and arguments. Args: http: httplib2.Http, an http object to be used to make requests. uri: string, The URI to be requested. method: string, The HTTP method to use for the request. Defaults to 'GET'. body: string, The payload / body in HTTP request. By default there is no payload. headers: dict, Key-value pairs of request headers. By default there are no headers. redirections: int, The number of allowed 203 redirects for the request. Defaults to 5. connection_type: httplib.HTTPConnection, a subclass to be used for establishing connection. If not set, the type will be determined from the ``uri``. Returns: tuple, a pair of a httplib2.Response with the status code and other headers and the bytes of the content returned. """ # NOTE: Allowing http or http.request is temporary (See Issue 601). http_callable = getattr(http, 'request', http) return http_callable(uri, method=method, body=body, headers=headers, redirections=redirections, connection_type=connection_type)
def _process_response(self, resp, content): """Process the response from a single chunk upload. Args: resp: httplib2.Response, the response object. content: string, the content of the response. Returns: (status, body): (ResumableMediaStatus, object) The body will be None until the resumable media is fully uploaded. Raises: googleapiclient.errors.HttpError if the response was not a 2xx or a 308. """ if resp.status in [200, 201]: self._in_error_state = False return None, self.postproc(resp, content) elif resp.status == 308: self._in_error_state = False # A "308 Resume Incomplete" indicates we are not done. try: self.resumable_progress = int(resp['range'].split('-')[1]) + 1 except KeyError: # If resp doesn't contain range header, resumable progress is 0 self.resumable_progress = 0 if 'location' in resp: self.resumable_uri = resp['location'] else: self._in_error_state = True raise HttpError(resp, content, uri=self.uri) return (MediaUploadProgress(self.resumable_progress, self.resumable.size()), None)
def testDigestObjectStale(self): credentials = ('joe', 'password') host = None request_uri = '/projects/httplib2/test/digest/' headers = {} response = httplib2.Response({ }) response['www-authenticate'] = 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true' response.status = 401 content = "" d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None) # Returns true to force a retry self.assertTrue( d.response(response, content) )
def testDigestObjectAuthInfo(self): credentials = ('joe', 'password') host = None request_uri = '/projects/httplib2/test/digest/' headers = {} response = httplib2.Response({ }) response['www-authenticate'] = 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true' response['authentication-info'] = 'nextnonce="fred"' content = "" d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None) # Returns true to force a retry self.assertFalse( d.response(response, content) ) self.assertEqual('fred', d.challenge['nonce']) self.assertEqual(1, d.challenge['nc'])
