本文整理汇总了Python中requests.Response类的典型用法代码示例。如果您正苦于以下问题:Python Response类的具体用法?Python Response怎么用?Python Response使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Response类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: get_http_request
def get_http_request(url, payload, method='POST', headers=None, use_proxy=False, use_proxy_auth=False, trust_env=True):
try:
session = Session()
session.trust_env = trust_env
session.proxies = Util.get_proxies() if use_proxy else None
session.auth = Util.get_proxy_auth() if use_proxy_auth else None
request = Request(
'POST' if method not in ('GET', 'POST') else method,
url,
data=payload if method == 'POST' else None,
params=payload if method == 'GET' else None,
headers=headers
)
prepped = request.prepare()
response = session.send(
prepped,
timeout=app.config['HTTP_REQUESTS_TIMEOUT']
)
session.close()
except Exception, e:
response = Response()
response.raise_for_status()
return response, 'Error al realizar la consulta - Motivo: {}'.format(e.message)
开发者ID:stabora,项目名称:nbsf,代码行数:27,代码来源:util.py
示例2: create_response
def create_response(**kw):
r = Response()
r._content = b''
r.status = 200
for k, v in kw.items():
setattr(r, k, v)
return r
开发者ID:cardforcoin,项目名称:chain-bitcoin-python,代码行数:7,代码来源:mock_http_adapter.py
示例3: test_reset_rabbit_connection_errors
def test_reset_rabbit_connection_errors():
rabbit_manager = Mock()
with patch('nameko.testing.utils.get_rabbit_connections') as connections:
connections.return_value = [{
'vhost': 'vhost_name',
'name': 'connection_name'
}]
# 500 error
response_500 = Response()
response_500.status_code = 500
error_500 = HTTPError(response=response_500)
rabbit_manager.delete_connection.side_effect = error_500
with pytest.raises(HTTPError):
reset_rabbit_connections("vhost_name", rabbit_manager)
# 404 error
response_404 = Response()
response_404.status_code = 404
error_404 = HTTPError(response=response_404)
rabbit_manager.delete_connection.side_effect = error_404
# does not raise
reset_rabbit_connections("vhost_name", rabbit_manager)
开发者ID:SivagnanamCiena,项目名称:nameko,代码行数:27,代码来源:test_utils.py
示例4: test_download_fails_due_to_hash_mismatch
def test_download_fails_due_to_hash_mismatch(self):
snap_content = b'1234567890'
mock_details = self.mock_get.return_value
mock_details.ok = True
mock_details.content = json.dumps({
'_embedded': {
'clickindex:package': [{
'download_url': 'http://localhost',
'anon_download_url': 'http://localhost',
'download_sha512': '12345',
}],
}
}).encode('utf-8')
mock_snap = Response()
mock_snap.status_code = 200
mock_snap._content = snap_content
self.mock_get.side_effect = [mock_details, mock_snap]
with self.assertRaises(RuntimeError) as raised:
download('os', 'edge', 'os.snap', None, 'amd64')
self.assertEqual("Failed to download 'os'", str(raised.exception))
self.mock_logger.info.assert_has_calls([
call("Getting details for 'os'"),
call("Downloading 'os'")])
self.assertTrue(os.path.exists('os.snap'))
开发者ID:0-T-0,项目名称:snapcraft,代码行数:27,代码来源:test_storeapi_download.py
示例5: test_get_token_with_scope
def test_get_token_with_scope(self, get_token):
the_response = Response()
the_response.status_code = 200
the_response.json = MagicMock(return_value={'access_token': 'foobar'})
get_token.return_value = the_response
block = OAuthBlock()
self.configure_block(block, {})
token = block.get_access_token(
'user', 'pass',
scope='my-scope',
addl_params={
'extra': 'value'
})
self.assertEqual(token['access_token'], 'foobar')
self.assertEqual(block.get_access_token_headers(), {
'Authorization': 'Bearer foobar'
})
get_token.assert_called_once_with(
'http://oauthbase/token',
data={
'username': 'user',
'password': 'pass',
'scope': 'my-scope',
'grant_type': 'password',
'extra': 'value'
})
开发者ID:nio-blocks,项目名称:oauth2_mixin,代码行数:29,代码来源:test_password.py
示例6: did_receive_response
async def did_receive_response(
self,
response: requests.Response,
request: Request
) -> typing.Any:
response.raise_for_status()
return response.json(object_hook=self.convert_to_object)
开发者ID:rmyers,项目名称:cannula,代码行数:7,代码来源:http.py
示例7: make_response
def make_response(self, status_code=200, reason='OK', data=None):
data = data or {}
response = Response()
response.status_code = status_code
response.reason = reason
response._content = json.dumps(data).encode('utf-8')
return response
开发者ID:0-T-0,项目名称:snapcraft,代码行数:7,代码来源:test_storeapi_login.py
示例8: __init__
def __init__(self):
"""Use response.text have performance problem, use response.html fix it.
This is an issue, maybe some init charset not set, requests use chardet
library to detect charset of the html content.
"""
ResponseLib.__init__(self)
开发者ID:FFng,项目名称:yascrapy,代码行数:7,代码来源:response_queue.py
示例9: test_token
def test_token(self, mock_head, mock_post, mock_service_creds):
# mock out the token request response
the_response = Response()
the_response.status_code = 200
the_response.json = MagicMock(return_value={'access_token': 'foobar'})
mock_post.return_value = the_response
block = OAuthBlock()
self.configure_block(block, {
"key_config_file": "tests/KEY_FILE_GOES_HERE.json"
})
token = block.get_access_token(
'https://www.googleapis.com/auth/analytics.readonly')
self.assertEqual(token['access_token'], 'foobar')
# Oauth2ServiceAccount.get_access_token is mocked, so
# Oauth2ServiceAccount.parse_token_from_response(r) is never called,
# and self._oauth_token is never assigned.
# Need to assign it here before checking block.get_access_token_headers
block._oauth_token = {'access_token': 'foobar'}
self.assertEqual(block.get_access_token_headers(), {
'Authorization': 'Bearer foobar'
})
开发者ID:nio-blocks,项目名称:oauth2_mixin,代码行数:26,代码来源:test_service_account.py
示例10: get_acacess_token_app
def get_acacess_token_app(tenant_id,client_id,client_secret,resource):
initialise()
url = "https://login.windows.net/" + tenant_id + "/oauth2/token"
body_data = "&grant_type=client_credentials&resource="+ resource +"&client_id="+ client_id + "&client_secret="+ urllib.quote_plus(client_secret)
headers = {"Content-Type":"application/x-www-form-urlencoded"}
req = Request(method="POST",url=url,data=body_data)
req_prepped = req.prepare()
s = Session()
res = Response()
res = s.send(req_prepped)
access_token_det = {}
if (res.status_code == 200):
responseJSON = json.loads(res.content)
access_token_det["details"]= responseJSON["access_token"]
access_token_det["status"]="1"
access_token_det["exp_time"]=responseJSON["expires_in"]
access_token_det["exp_date"]=responseJSON["expires_on"]
access_token_det["accessDetails"]=responseJSON
else:
access_token_det["details"]= str(res.status_code) + str(res.json())
access_token_det["status"]="0"
return access_token_det
开发者ID:shwetams,项目名称:arm-samples-py,代码行数:25,代码来源:views.py
示例11: parse_arrivals_from_infotrafic
def parse_arrivals_from_infotrafic(line_id: int, stations: Dict[str, Station], response: requests.Response, include_unknown_stations: bool = False) -> Tuple[List[Tuple[Union[Station,str], Arrival]]]:
response.raise_for_status()
if response.status_code == requests.codes.ok:
bs = bs4.BeautifulSoup(response.text, "html.parser")
prevcolor = None
datacolor = '00BFFF'
routes = []
route = None
tz = pytz.timezone("Europe/Bucharest")
now = tzlocal.get_localzone().localize(datetime.now()).astimezone(tz).replace(second=0, microsecond=0)
for row in bs.find_all("table"):
if row['bgcolor'] == datacolor:
if prevcolor != datacolor:
route = []
routes.append(route)
cols = row.find_all("b")
raw_station_name = cols[1].text.strip()
station = stations.get(raw_station_name, None)
arrival = parse_arrival(now, line_id, station.station_id if station else -1, cols[2].text)
if station is not None or include_unknown_stations:
route.append((station if station is not None else raw_station_name, arrival))
prevcolor = row['bgcolor']
return routes if route else None
return None
开发者ID:HackTM2016,项目名称:CityTransportTM,代码行数:28,代码来源:ratt.py
示例12: test_get_json_data
def test_get_json_data(self, gobase_get_full_response):
response = Response()
response._content = "not a json object"
gobase_get_full_response.return_value = response
with self.assertRaises(GoCdApiException):
self.gobase.get_json_data(self.gobase.url)
开发者ID:joaogbcravo,项目名称:gocdapi,代码行数:7,代码来源:test_gobase.py
示例13: logged_in
def logged_in(self, login='fred', travis_user=True):
""" A context manager to do stuff, while logged in as fred. """
response = Response()
response._content = json.dumps(
dict(id=12345, login=login, name='Fred')
)
data = Mock()
data.get = Mock(return_value=response)
data.access_token = GH_TOKEN
true = Mock(return_value=True)
if travis_user:
travis_patch = patch('travis_utils.is_travis_user', true)
travis_patch.start()
with patch('statiki.github', Mock(spec=OAuth2Service)) as gh:
gh.get_auth_session = Mock(return_value=data)
try:
yield self.app.get('/authorized?code="bazooka"')
finally:
if travis_user:
travis_patch.stop()
self.app.get('/logout')
开发者ID:punchagan,项目名称:statiki,代码行数:28,代码来源:test_statiki.py
示例14: test_yahoo_get_by_date
def test_yahoo_get_by_date(yahoo, logger):
sample = Response()
sample.status_code = 200
sample._content = YAHOO_RESPONSE
yahoo._get = lambda *a, **kw: sample
assert yahoo.get_by_date(date.today(), "CZK", logger) == Decimal("25.959")
开发者ID:business-factory,项目名称:gold-digger,代码行数:7,代码来源:test_yahoo.py
示例15: build_response
def build_response(request,
status_code=200,
headers={},
content='(none)'):
"""
Build a :class:`requests.Response` object on the basis of the passed
parameters.
"""
response = Response()
response.status_code = status_code
response.reason = responses[status_code]
response.headers = CaseInsensitiveDict(headers)
# Pretend that we've already read from the socket
response._content = content
response.encoding = get_encoding_from_headers(response.headers)
response.url = request.url
response.raw = MockRawResponse()
# Give the Response some context.
response.request = request
response.connection = MockConnection()
return response
开发者ID:alphagov,项目名称:ghtools,代码行数:26,代码来源:requestmocker.py
示例16: test_add_channel_channel_already_exists
def test_add_channel_channel_already_exists(mock_staff_client, patched_users_api):
"""Channel already exists with that channel name"""
response_409 = Response()
response_409.status_code = statuses.HTTP_409_CONFLICT
mock_staff_client.channels.create.return_value = response_409
title = "title"
name = "name"
description = "public description"
channel_type = "private"
input_search = Search.from_dict({"unmodified": "search"})
role = RoleFactory.create()
mod = UserFactory.create()
with pytest.raises(ChannelAlreadyExistsException):
api.add_channel(
original_search=input_search,
title=title,
name=name,
description=description,
channel_type=channel_type,
program_id=role.program.id,
creator_id=mod.id,
)
mock_staff_client.channels.create.assert_called_once_with(
title=title,
name=name,
description=description,
channel_type=channel_type,
)
开发者ID:mitodl,项目名称:micromasters,代码行数:31,代码来源:api_test.py
示例17: test_private_user
def test_private_user(self, mock_get, mock_id, mock_auth, mock_retry):
blk = InstagramSearchByUser()
self.configure_block(blk, {
"queries": [
"user1",
"user2"
]
})
blk.queries = ['1', '2']
blk._n_queries = len(blk.queries())
resp = Response()
resp.status_code = 400
resp.json = Mock()
resp.json.return_value = \
{
'meta': {
'error_type': 'APINotAllowedError',
'code': 400,
'error_message': 'you cannot view this resource'
}
}
mock_get.return_value = resp
paging = False
self.assertEqual(0, blk._idx)
blk.poll(paging)
# skip to next idx because we are not retrying.
self.assertEqual(1, blk._idx)
开发者ID:nio-blocks,项目名称:instagram,代码行数:27,代码来源:tests_instagram_search_by_user_block.py
示例18: test_snap_already_downloaded
def test_snap_already_downloaded(self):
snap_content = b'1234567890'
snap_sha512 = ('12b03226a6d8be9c6e8cd5e55dc6c7920caaa39df14aab92d5e'
'3ea9340d1c8a4d3d0b8e4314f1f6ef131ba4bf1ceb9186ab87c'
'801af0d5c95b1befb8cedae2b9')
with open('os.snap', 'wb') as f:
f.write(snap_content)
mock_details = self.mock_get.return_value
mock_details.ok = True
mock_details.content = json.dumps({
'download_url': 'http://localhost',
'download_sha512': snap_sha512,
}).encode('utf-8')
mock_snap = Response()
mock_snap.status_code = 200
mock_snap._content = snap_content
self.mock_get.side_effect = [mock_details, mock_snap]
download('os', 'os.snap', None, 'amd64')
self.mock_logger.info.assert_has_calls([
call("Getting details for 'os'"),
call("Already downloaded 'os'")])
self.assertTrue(os.path.exists('os.snap'))
开发者ID:fallen,项目名称:snapcraft,代码行数:27,代码来源:test_storeapi_download.py
示例19: test_retry
def test_retry(self, mock_get, mock_id, mock_auth, mock_retry):
blk = InstagramSearchByUser()
self.configure_block(blk, {
"queries": [
"user1",
"user2"
]
})
blk.queries = ['1', '2']
blk._n_queries = len(blk.queries())
resp = Response()
resp.status_code = 400
resp.json = Mock()
resp.json.return_value = \
{
'meta': {
'error_type': 'WardrobeMalfunction',
'code': 400,
'error_message': "Your pants don't fit"
}
}
mock_get.return_value = resp
paging = False
self.assertEqual(0, blk._idx)
blk.poll(paging)
# don't skip to next idx because we are retrying.
self.assertEqual(0, blk._idx)
开发者ID:nio-blocks,项目名称:instagram,代码行数:27,代码来源:tests_instagram_search_by_user_block.py
示例20: test_upload_app_ok
def test_upload_app_ok(self):
# fake upload response
mock_response = self.mock_post.return_value
mock_response.ok = True
mock_response.return_value = {
'success': True,
'status_url': 'http://example.com/status/',
}
# fake poll status response
application_url = 'http://example.com/app/1'
revision = '1'
ok_response = Response()
ok_response.status_code = 200
ok_response.encoding = 'utf-8'
ok_response._content = json.dumps(
{'completed': True, 'revision': revision,
'application_url': application_url}).encode('utf-8')
self.mock_get.return_value = ok_response
success = upload(self.binary_file.name, 'foo')
self.assertTrue(success)
self.assertIn(
call('Application uploaded successfully (as revision {})'.format(
revision)),
self.mock_logger.info.call_args_list)
self.assertIn(call('Please check out the application at: %s\n',
application_url),
self.mock_logger.info.call_args_list)
开发者ID:0-T-0,项目名称:snapcraft,代码行数:29,代码来源:test_storeapi_upload.py
注:本文中的requests.Response类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论