本文整理汇总了Python中responses.add_callback函数的典型用法代码示例。如果您正苦于以下问题:Python add_callback函数的具体用法?Python add_callback怎么用?Python add_callback使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了add_callback函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_intent_verified_returns_verified
def test_intent_verified_returns_verified(client):
responses.add_callback(
responses.GET, CALLBACK_URL, callback=verify_intent_request_callback
)
result = intent_verified("subscribe", TOPIC_URL, CALLBACK_URL, 10)
assert result is True
开发者ID:DBeath,项目名称:flask-feedrsub,代码行数:7,代码来源:websub_handler_test.py
示例2: test_calc_api
def test_calc_api():
def request_callback(request):
payload = json.loads(request.body)
resp_body = {'value': sum(payload['numbers'])}
headers = {'request-id': '728d329e-0e86-11e4-a748-0c84dc037c13'}
return (200, headers, json.dumps(resp_body))
responses.add_callback(
responses.POST, 'http://calc.com/sum',
callback=request_callback,
content_type='application/json',
)
resp = requests.post(
'http://calc.com/sum',
json.dumps({'numbers': [1, 2, 3]}),
headers={'content-type': 'application/json'},
)
assert resp.json() == {'value': 6}
assert len(responses.calls) == 1
assert responses.calls[0].request.url == 'http://calc.com/sum'
assert responses.calls[0].response.text == '{"value": 6}'
assert (
responses.calls[0].response.headers['request-id'] ==
'728d329e-0e86-11e4-a748-0c84dc037c13'
)
开发者ID:macrotim,项目名称:tutorials-applied,代码行数:29,代码来源:test_tutorial3.py
示例3: test_assignee_search
def test_assignee_search(self):
responses.add(
responses.GET,
'https://example.atlassian.net/rest/api/2/project',
json=[{'key': 'HSP', 'id': '10000'}],
match_querystring=False
)
def responder(request):
query = parse_qs(urlparse(request.url).query)
assert 'HSP' == query['project'][0]
assert 'bob' == query['query'][0]
return (200, {}, SAMPLE_USER_SEARCH_RESPONSE)
responses.add_callback(
responses.GET,
'https://example.atlassian.net/rest/api/2/user/assignable/search',
callback=responder,
content_type='json',
match_querystring=False
)
org = self.organization
self.login_as(self.user)
path = reverse('sentry-extensions-jira-search', args=[org.slug, self.integration.id])
resp = self.client.get('%s?project=10000&field=assignee&query=bob' % (path,))
assert resp.status_code == 200
assert resp.data == [
{'value': 'deadbeef123', 'label': 'Bobby - [email protected]'}
]
开发者ID:yaoqi,项目名称:sentry,代码行数:31,代码来源:test_search_endpoint.py
示例4: set_up_glove
def set_up_glove(url: str, byt: bytes, change_etag_every: int = 1000):
# Mock response for the datastore url that returns glove vectors
responses.add(
responses.GET,
url,
body=byt,
status=200,
content_type='application/gzip',
stream=True,
headers={'Content-Length': str(len(byt))}
)
etags_left = change_etag_every
etag = "0"
def head_callback(_):
"""
Writing this as a callback allows different responses to different HEAD requests.
In our case, we're going to change the ETag header every `change_etag_every`
requests, which will allow us to simulate having a new version of the file.
"""
nonlocal etags_left, etag
headers = {"ETag": etag}
# countdown and change ETag
etags_left -= 1
if etags_left <= 0:
etags_left = change_etag_every
etag = str(int(etag) + 1)
return (200, headers, "")
responses.add_callback(
responses.HEAD,
url,
callback=head_callback
)
开发者ID:Jordan-Sauchuk,项目名称:allennlp,代码行数:34,代码来源:file_utils_test.py
示例5: test_update_with_invalid_mentenanceid
def test_update_with_invalid_mentenanceid(self):
def request_callback(request):
method = json.loads(request.body)['method']
if method == 'maintenance.get':
return (200, {}, json.dumps(get_response('22')))
else:
return (200, {}, json.dumps(update_response('22')))
responses.add(
responses.POST, 'https://example.com/zabbix/api_jsonrpc.php',
body=json.dumps({
'result': 'authentication_token',
'id': 1,
'jsonrpc': '2.0'
}),
status=200,
content_type='application/json'
)
m = self._makeOne(
host='https://example.com',
user='osamunmun',
password='foobar')
maintenance_id = '22'
responses.reset()
responses.add_callback(responses.POST, 'https://example.com/zabbix/api_jsonrpc.php',
callback=request_callback,
content_type='application/json')
for key in ('maintenanceid', 'name', 'active_since', 'active_till'):
required_params = {'maintenanceid': maintenance_id,
'name': 'test',
'active_since': '2004-04-01T12:00+09:00',
'active_till': '2014-04-01T12:00+09:00'}
with self.assertRaises(jsonschema.exceptions.ValidationError):
required_params.pop(key)
m.update(required_params)
开发者ID:osamunmun,项目名称:zabbix-api-client,代码行数:35,代码来源:test_maintenance.py
示例6: run
def run():
responses.add_callback(responses.GET, url, request_callback)
resp = requests.get(url)
assert resp.text == "test callback"
assert resp.status_code == status
assert 'foo' in resp.headers
assert resp.headers['foo'] == 'bar'
开发者ID:MattBlack85,项目名称:responses,代码行数:7,代码来源:test_responses.py
示例7: test_process
def test_process(self):
def callback(request):
payload = dict(parse_qsl(request.body))
success = True
if payload['request_id'] != 1:
success = False
success = all(i in payload for i in ['ext_auth_success_uri', 'ext_auth_fail_uri',
'request_token', 'instance_id'])
if success:
return 200, {}, json.dumps({'status': 'success', 'invoice_id': '0'})
return 200, {}, json.dumps({'status': 'refused', 'error': 'illegal_params'})
responses.add_callback(
responses.POST, re.compile('https?://.*/api/process-external-payment'),
callback=callback,
content_type='application/json',
)
resp = self.api.process({'request_id': 1,
'ext_auth_success_uri': 'test',
'ext_auth_fail_uri': 'test_fail',
'request_token': 'request_token',
'instance_id': '123',
})
self.assertEqual(resp.status, 'success')
self.assertEqual(resp.invoice_id, '0')
self.assertEqual(len(responses.calls), 2)
开发者ID:Krukov,项目名称:yandex-money-sdk-python,代码行数:28,代码来源:testWallet.py
示例8: test_PushoverAPI_gets_group_info
def test_PushoverAPI_gets_group_info(PushoverAPI):
"""Test getting group info"""
url_re = re.compile('https://api\.pushover\.net/1/groups/g[a-zA-Z0-9]*\.json')
responses.add_callback(
responses.GET,
url_re,
callback=groups_callback,
content_type='application/json'
)
resp = PushoverAPI.group_info(TEST_GROUP)
# request_body = parse_qs(resp.request.body)
# assert request_body['token'][0] == TEST_TOKEN
# assert resp.request.path_url.split('/')[-1].split('.')[0] == TEST_GROUP
assert resp == {
'status': 1,
'request': TEST_REQUEST_ID,
'name': TEST_GROUP_NAME,
'users': [
{
'user': TEST_USER,
'device': TEST_DEVICES[0],
'memo': '',
'disabled': False
},
{
'user': TEST_USER,
'device': TEST_DEVICES[1],
'memo': '',
'disabled': False
}
]
}
开发者ID:scolby33,项目名称:pushover_complete,代码行数:34,代码来源:test_PushoverAPI.py
示例9: test_PushoverAPI_sends_complex_message
def test_PushoverAPI_sends_complex_message(PushoverAPI):
"""Test sending a more complex message."""
responses.add_callback(
responses.POST,
urljoin(PUSHOVER_API_URL, 'messages.json'),
callback=messages_callback,
content_type='application/json'
)
resp = PushoverAPI.send_message(
TEST_USER,
TEST_MESSAGE,
device=TEST_DEVICES[0],
title=TEST_TITLE,
url=TEST_URL,
url_title=TEST_URL_TITLE,
priority=1,
timestamp=100,
sound='gamelan'
)
# request_body = parse_qs(resp.request.body)
# assert request_body['token'][0] == TEST_TOKEN
# assert request_body['user'][0] == TEST_USER
# assert request_body['device'][0] == TEST_DEVICES[0]
# assert request_body['title'][0] == TEST_TITLE
# assert request_body['url'][0] == TEST_URL
# assert request_body['url_title'][0] == TEST_URL_TITLE
# assert int(request_body['priority'][0]) == 1
# assert int(request_body['timestamp'][0]) == 100
# assert request_body['sound'][0] == 'gamelan'
assert resp == {
'status': 1,
'request': TEST_REQUEST_ID
}
开发者ID:scolby33,项目名称:pushover_complete,代码行数:34,代码来源:test_PushoverAPI.py
示例10: test_PushoverAPI_sends_multiple_simple_messages
def test_PushoverAPI_sends_multiple_simple_messages(PushoverAPI):
"""Test sending multiple simple messages through one HTTP session."""
responses.add_callback(
responses.POST,
urljoin(PUSHOVER_API_URL, 'messages.json'),
callback=messages_callback,
content_type='application/json'
)
messages = [{
'user': TEST_USER,
'message': TEST_MESSAGE
}] * 3
resps = PushoverAPI.send_messages(messages)
# request_bodies = [parse_qs(resp.request.body) for resp in resps]
# assert len(resps) == 3
# assert all(request_body['token'][0] == TEST_TOKEN for request_body in request_bodies)
# assert all(request_body['user'][0] == TEST_USER for request_body in request_bodies)
# assert all(request_body['message'][0] == TEST_MESSAGE for request_body in request_bodies)
# assert all(request_body['html'][0] == 'False' for request_body in request_bodies)
assert all(resp == {
'status': 1,
'request': TEST_REQUEST_ID
} for resp in resps)
开发者ID:scolby33,项目名称:pushover_complete,代码行数:25,代码来源:test_PushoverAPI.py
示例11: test_PushoverAPI_gets_receipt
def test_PushoverAPI_gets_receipt(PushoverAPI):
"""Test the retrieval of receipt details."""
url_re = re.compile('https://api\.pushover\.net/1/receipts/r[a-zA-Z0-9]*\.json')
responses.add_callback(
responses.GET,
url_re,
callback=receipt_callback,
content_type='application/json'
)
resp = PushoverAPI.check_receipt(TEST_RECEIPT_ID)
# request_body = parse_qs(resp.request.body)
# assert request_body['token'][0] == TEST_TOKEN
# assert resp.request.path_url.split('/')[-1].split('.')[0] == TEST_RECEIPT_ID
assert resp == {
'status': 1,
'request': TEST_REQUEST_ID,
'acknowledged': 1,
'acknowledged_at': 100,
'acknowledged_by': TEST_USER,
'acknowledged_by_device': TEST_DEVICES[0],
'last_delivered_at': 100,
'expired': 1,
'expires_at': 100,
'called_back': 0,
'called_back_at': 100
}
开发者ID:scolby33,项目名称:pushover_complete,代码行数:27,代码来源:test_PushoverAPI.py
示例12: httpShouldReturn
def httpShouldReturn(self, body=None, callback=None, scheme='http', host='test.wikipedia.org', path='/w/',
script='api'):
url = '{scheme}://{host}{path}{script}.php'.format(scheme=scheme, host=host, path=path, script=script)
if body is None:
responses.add_callback(responses.POST, url, callback=callback, content_type='application/json')
else:
responses.add(responses.POST, url, body=body, content_type='application/json')
开发者ID:fhocutt,项目名称:mwclient,代码行数:7,代码来源:test_client.py
示例13: test_create
def test_create(self):
# Post call ###########################################################
def request_callback(request):
payload = json.loads(request.body)
assert 'OData-Version' in request.headers, 'OData-Version header not in request'
assert 'ProductID' not in payload, 'Payload contains primary key'
assert '@odata.type' in payload, 'Payload did not contain @odata.type'
payload['ProductID'] = 1
resp_body = payload
headers = {}
return requests.codes.created, headers, json.dumps(resp_body)
responses.add_callback(
responses.POST, Product.__odata_url__(),
callback=request_callback,
content_type='application/json',
)
#######################################################################
new_product = Product()
new_product.name = u'New Test Product'
new_product.category = u'Category #1'
new_product.price = 34.5
Service.save(new_product)
assert new_product.id is not None, 'Product.id is not set'
开发者ID:bpowell65536,项目名称:python-odata,代码行数:31,代码来源:test_objects.py
示例14: test_read
def test_read(self):
expected_id = 1024
expected_name = 'Existing entity'
expected_category = 'Existing category'
expected_price = Decimal('85.2')
# Initial data ########################################################
def request_callback(request):
payload = {
'ProductID': expected_id,
'ProductName': expected_name,
'Category': expected_category,
'Price': float(expected_price),
}
resp_body = {'value': [payload]}
headers = {}
return requests.codes.ok, headers, json.dumps(resp_body)
responses.add_callback(
responses.GET, Product.__odata_url__(),
callback=request_callback,
content_type='application/json',
)
#######################################################################
product = Service.query(Product).first()
assert product.id == expected_id
assert product.name == expected_name
assert product.category == expected_category
assert product.price == expected_price
开发者ID:bpowell65536,项目名称:python-odata,代码行数:31,代码来源:test_objects.py
示例15: test_lookup_episode
def test_lookup_episode():
responses.add_callback(
responses.GET, 'http://mock/search',
callback=search_callback,
content_type='application/json'
)
Trakt.base_url = 'http://mock'
episode = Trakt['search'].lookup('tt0959621', 'imdb')
assert episode.keys == [
(1, 1),
('tvdb', '349232'),
('tmdb', '62085'),
('imdb', 'tt0959621'),
('tvrage', '637041'),
('trakt', '73482')
]
assert episode.title == "Pilot"
assert sorted(episode.images.keys()) == ['screenshot']
assert episode.overview is not None
assert episode.score is None
assert episode.show.keys == [
('slug', 'breaking-bad'),
('trakt', '1388')
]
assert episode.show.title == "Breaking Bad"
assert episode.show.year == 2008
assert sorted(episode.show.images.keys()) == ['fanart', 'poster']
开发者ID:TWaalen,项目名称:trakt.py,代码行数:35,代码来源:test_search.py
示例16: test_send
def test_send(self):
with app.app_context():
responses.add(responses.GET, '{}/foo1'.format(self.rmq_base_url), status=200)
self.assertEqual(send('get', 'foo1').status_code, 200)
#
# timeouts and other requests errors
#
def callback(request):
raise requests.RequestException('Requests exception')
responses.add_callback(
responses.GET, '{}/foo2'.format(self.rmq_base_url), callback=callback,
)
with self.assertRaises(InternalServerError):
send('get', 'foo2')
#
# Test non-200 resposes with raise_for_status=True
#
responses.add(
responses.GET, '{}/foo3'.format(self.rmq_base_url), status=400,
)
with self.assertRaises(InternalServerError):
send('get', 'foo3')
#
# Test that we return a resposes with raise_for_status=False
#
responses.add(
responses.GET, '{}/foo4'.format(self.rmq_base_url), status=400,
)
response = send('get', 'foo4', raise_for_status=False)
self.assertEqual(response.status_code, 400)
开发者ID:carriercomm,项目名称:rabbitmqapi,代码行数:34,代码来源:tests.py
示例17: test_import_with_too_many_redirects
def test_import_with_too_many_redirects(client, user):
csv_url = "http://example.com/file.csv"
def request_callback(request):
headers={
'HTTP/1.1': '301 Moved Permanently',
'Location': csv_url,
}
return (301, headers, '')
# fetch csv mock response
responses.add_callback(
responses.GET,
csv_url,
callback=request_callback,
)
resp = client.post_json(
reverse('import-csv-from-link'),
params=dict(url=csv_url),
headers=[('Authorization', str('Token %s' % user.auth_token))],
status=400
)
assert 'Exceeded 30 redirects.' == resp.json['url'][0]
开发者ID:dincamihai,项目名称:csvapi,代码行数:25,代码来源:test_import.py
示例18: registration
def registration(uri):
"""Responses handler registration.
Registers a handler for a given URI with Responses
so that it can be intercepted and handed to
Stack-In-A-Box.
:param uri: URI used for the base of the HTTP requests
:returns: n/a
"""
# log the URI that is used to access the Stack-In-A-Box services
logger.debug('Registering Stack-In-A-Box at {0} under Python Responses'
.format(uri))
# tell Stack-In-A-Box what URI to match with
StackInABox.update_uri(uri)
# Build the regex for the URI and register all HTTP verbs
# with Responses
regex = re.compile('(http)?s?(://)?{0}:?(\d+)?/'.format(uri),
re.I)
METHODS = [
responses.DELETE,
responses.GET,
responses.HEAD,
responses.OPTIONS,
responses.PATCH,
responses.POST,
responses.PUT
]
for method in METHODS:
responses.add_callback(method,
regex,
callback=responses_callback)
开发者ID:BenjamenMeyer,项目名称:stackInABox,代码行数:35,代码来源:core.py
示例19: setUp
def setUp(self):
self.api = ExternalPayment(self.client_id)
responses.add_callback(
responses.POST, re.compile('https?://.*/api/instance-id'),
callback=self.request_callback,
content_type='application/json',
)
开发者ID:Krukov,项目名称:yandex-money-sdk-python,代码行数:7,代码来源:testWallet.py
示例20: mock_login_response
def mock_login_response(self, status):
""" Mock the response of the marketing site login """
response_url = '{root}/users/{username}'.format(
root=self.api_root,
username=self.username
)
def request_callback(request): # pylint: disable=unused-argument
headers = {
'location': response_url
}
return (302, headers, None)
responses.add_callback(
responses.POST,
'{root}/user'.format(root=self.api_root),
callback=request_callback,
content_type='text/html',
)
responses.add(
responses.GET,
response_url,
body='',
content_type='text/html',
status=status
)
开发者ID:edx,项目名称:course-discovery,代码行数:27,代码来源:mixins.py
注:本文中的responses.add_callback函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论