Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_entity_get_save(self):
with open('tests/fixtures/xml/advertiser.xml') as f:
fixture = f.read()
responses.add(responses.GET, 'https://api.mediamath.com/api/v2.0/advertisers/1',
body=fixture,
content_type='application/xml')
responses.add(responses.POST, 'https://api.mediamath.com/api/v2.0/advertisers/1',
body=fixture,
content_type='application/xml')
adv = self.t1.get('advertisers', 1)
assert adv.id == 1, "Expected ID 1, got: %d" % adv.id
assert all(
hasattr(adv, item) for item in [
'id',
'name',
'status',
'agency_id',
'created_on',
'updated_on',
'ad_server_id',
]
), 'Expected a full record, got: %r' % adv
adv.save()
def test_input_error(self):
responses.add(responses.POST, 'https://api.smartystreets.com/street-address',
body='', status=400,
content_type='application/json')
self.assertRaises(SmartyStreetsInputError, self.client.street_addresses, [{}, {}])
def test_name_already_registered(self):
payload = {"error_list": [{"code": "already_registered"}]}
responses.add(responses.POST, self.api_url, json=payload, status=409)
response = self.client.post(self.endpoint_url, data=self.data)
errors = response.json["errors"]
self.assertStatus(response, 409)
self.assertEqual(len(errors), 1)
self.assertEqual(errors[0]["code"], "already_registered")
def test_get_versions(self):
def create_callback(request, snoop):
snoop["payload"] = request.body
return 200, {}, "\n".join(json.dumps(c) for c in self._versions_json)
p = Project.from_json(self.tamr, self._project_json).as_mastering()
post_url = f"http://localhost:9100/api/versioned/v1/{p.api_path}/publishedClusterVersions"
snoop = {}
responses.add_callback(
responses.POST, post_url, partial(create_callback, snoop=snoop)
)
clusters = list(p.published_cluster_versions(self._cluster_ids))
expected_clusters = [PublishedCluster(c) for c in self._versions_json]
self.assertEqual(
snoop["payload"], "\n".join([json.dumps(i) for i in self._cluster_ids])
)
self.assertEqual(len(clusters), len(expected_clusters))
for actual, expected in zip(clusters, expected_clusters):
self.assertEqual(actual.__repr__(), expected.__repr__())
self.assertEqual(len(actual.versions), len(expected.versions))
def mock_blenderid_validate_unhappy(self):
"""Sets up Responses to mock unhappy validation flow."""
responses.add(responses.POST,
'%s/u/validate_token' % self.app.config['BLENDER_ID_ENDPOINT'],
json={'status': 'fail'},
status=403)
response_data = {
"message": "feedback_required",
"spam": True,
"feedback_title": "Sorry, this feature isn't available right now",
"feedback_message": "An error occurred while processing this "
+ "request. Please try again later. We restrict certain content "
+ "and actions to protect our community. Tell us if you think we "
+ "made a mistake.",
"feedback_url": "repute/report_problem/instagram_comment/",
"feedback_appeal_label": "Report problem",
"feedback_ignore_label": "OK",
"feedback_action": "report_problem",
"status": "fail",
}
responses.add(
responses.POST,
"{api_url}media/{media_id}/comment/".format(
api_url=API_URL, media_id=media_id
),
json=response_data,
status=400,
)
assert not self.bot.comment(media_id, comment_txt)
"training_status": {
"objects": {
"in_progress": "true",
"data_changed": "true",
"ready": "true",
"latest_failed": "true",
"description": "description"
}
},
"created": "2000-01-23T04:56:07.000+00:00",
"name": "name",
"description": "description",
"image_count": 0,
"updated": "2000-01-23T04:56:07.000+00:00"
}
responses.add(responses.POST,
url,
body=json.dumps(response),
status=200,
content_type='application/json')
authenticator = IAMAuthenticator('bogusapikey')
service = ibm_watson.VisualRecognitionV4('YYYY-MM-DD',
authenticator=authenticator)
service.set_service_url(base_url)
detailed_response = service.update_collection(
collection_id='collection_id',
name='name',
description='description')
result = detailed_response.get_result()
assert result is not None
def test_directions_json(self):
query = deepcopy(ENDPOINTS_QUERIES[self.name]['directions'])
responses.add(
responses.POST,
'https://api.openrouteservice.org/v2/directions/{}/json'.format(query['profile']),
status=200,
json=ENDPOINTS_RESPONSES[self.name]['directions']['json'],
content_type='application/json'
)
routes = self.client.directions(**query, format='json')
query['coordinates'] = query['locations']
del query['locations']
del query['profile']
self.assertEqual(1, len(responses.calls))
self.assertEqual(query, json.loads(responses.calls[0].request.body.decode('utf-8')))
self.assertIsInstance(routes, Direction)
def test_can_end(self):
responses.add(
responses.POST, "http://foo.com/user/pending?api_key=KEY&api_secret=SECRET"
)
self.component.pending()