Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
status=200)
# AND registry server successfully finalizes upload
responses.add(
responses.POST, 'https://example.com/api/package/upload',
json={'status': 'queued'},
status=200)
# WHEN `dpm publish` is invoked
result = self.invoke(cli, ['publish'])
# THEN published package url should be printed to stdout
self.assertRegexpMatches(result.output, 'Datapackage successfully published. It is available at https://example.com/user/some-datapackage')
# AND 6 requests should be sent
self.assertEqual(
[(x.request.method, x.request.url)
for x in responses.calls],
[
# POST authorization
('POST', 'https://example.com/api/auth/token'),
# POST authorize presigned url for s3 upload
('POST', 'https://example.com/api/datastore/authorize'),
# POST data to s3
('POST', 'https://s3.fake/put_here_datapackege'),
('POST', 'https://s3.fake/put_here_readme'),
('POST', 'https://s3.fake/put_here_resource'),
# POST finalize upload
('POST', 'https://example.com/api/package/upload')
])
# AND exit code should be 0
self.assertEqual(result.exit_code, 0)
def test_headers_are_set_during_object_creation():
responses.add(responses.POST, 'http://test.com/graphql',
body='''
{"errors": [], "data": {"title": "blog title"}}''',
status=200,
content_type='application/json')
gql = Gql(api='http://test.com/graphql', default_headers={'foo': 'bar'})
gql.send(query='query {}', variables={})
assert len(responses.calls) == 1
assert responses.calls[0].request.url == 'http://test.com/graphql'
assert 'foo' in responses.calls[0].request.headers.keys()
def test_api_500_no_answer(self):
responses.add(
responses.Response(method="GET", url=self.api_url, status=500)
)
response = self.client.get(self.badge_url)
assert len(responses.calls) == 1
called = responses.calls[0]
assert called.request.url == self.api_url
assert response.status_code == 502
def test_elevation_line(self):
responses.add(responses.POST,
'https://api.openrouteservice.org/elevation/line',
json=self.valid_query,
status=200,
content_type='application/json')
resp = self.client.elevation_line(**self.valid_query)
self.assertEquals(len(responses.calls), 1)
self.assertEquals(resp, self.valid_query)
def test_send_finished_task(self):
responses.add(responses.POST, "http://example.com/")
config = {"url": "http://example.com/"}
self.notifier.send_deploy(
self.deploy, self.task, config, NotifierEvent.TASK_FINISHED
)
call = responses.calls[0]
assert len(responses.calls) == 1
assert responses.calls[0].request.url == "http://example.com/"
body = responses.calls[0].request.body
payload = json.loads(body)
assert payload
responses.add(
responses.POST,
message_url,
body=json.dumps(message_response),
status=200,
content_type='application/json')
message = assistant.message(
workspace_id=workspace_id,
input={'text': 'Turn on the lights'},
context=None).get_result()
assert message is not None
assert responses.calls[0].request.url == message_url1
assert 'x-watson-learning-opt-out' in responses.calls[0].request.headers
assert responses.calls[0].request.headers['x-watson-learning-opt-out'] == 'true'
assert responses.calls[0].response.text == json.dumps(message_response)
# test context
responses.add(
responses.POST,
message_url,
body=message_response,
status=200,
content_type='application/json')
message_ctx = {
'context': {
'conversation_id': '1b7b67c0-90ed-45dc-8508-9488bc483d5b',
'system': {
'dialog_stack': ['root'],
def test_request_made(self):
responses.add(responses.GET, 'https://example.com', status=200)
donation = Donation.objects.create(
timereceived=datetime.datetime(2018, 1, 1),
comment='',
amount=Decimal(1.5),
domain='PAYPAL',
donor=self.donor,
event=self.event,
)
eventutil.post_donation_to_postbacks(donation)
assert len(responses.calls) == 1
assert (
responses.calls[0].request.url
== 'https://example.com/?comment=&amount=1.5&timereceived=2018-01-01+00%3A00%3A00&donor__visibility=FIRST&domain=PAYPAL&id=1&donor__visiblename=%28No+Name%29'
)
assert responses.calls[0].response.status_code == 200
# pylint: disable=redefined-outer-name
responses.add(
responses.POST,
"{0}/Cars".format(service.url),
headers={'Content-type': 'application/json'},
json={'d': {
'Name': 'Hadraplan',
}},
status=201)
entity = {'Name': 'Hadraplan'}
result = service.entity_sets.Cars.create_entity().set(**entity).execute()
assert result.Name == 'Hadraplan'
assert_request_contains_header(responses.calls[0].request.headers, 'X-Requested-With', 'X')
def test_as_register():
endpoint = base_url + "register"
responses.add(responses.POST, endpoint, body="{}", status=200,
content_type="application_json")
resp1 = api.as_register("username")
resp2 = api.as_register("username", "sample_token")
assert resp1.json() == {}
assert resp2.json() == {}
assert len(responses.calls) == 2
assert responses.calls[0].request.url == endpoint + "?access_token=example_token"
assert responses.calls[1].request.url == endpoint + "?access_token=sample_token"
assert json.loads(responses.calls[0].request.body)["username"] == "username"
assert json.loads(responses.calls[1].request.body)["type"] == "m.login.application_service"
def test_key_sent(self):
responses.add(responses.GET,
"https://maps.googleapis.com/maps/api/geocode/json",
body='{"status":"OK","results":[]}',
status=200,
content_type="application/json")
client = googlemaps.Client(key="AIzaasdf")
client.geocode("Sesame St.")
self.assertEqual(1, len(responses.calls))
self.assertURLEqual("https://maps.googleapis.com/maps/api/geocode/json?"
"key=AIzaasdf&address=Sesame+St.",
responses.calls[0].request.url)