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_function_import_http_forbidden(service):
"""Function Imports report user friendly error message for Forbidden"""
# pylint: disable=redefined-outer-name
responses.add(
responses.GET,
"{0}/refresh".format(service.url),
status=403)
with pytest.raises(HttpError) as caught:
service.functions.refresh.execute()
assert str(caught.value) == 'Missing privileges to call Function Import refresh'
def _fetch_metadata(connection, url, logger):
# download metadata
logger.info('Fetching metadata')
resp = connection.get(url + '$metadata')
logger.debug('Retrieved the response:\n%s\n%s',
'\n'.join((f'H: {key}: {value}' for key, value in resp.headers.items())),
resp.content)
if resp.status_code != 200:
raise HttpError(
'Metadata request failed, status code: {}, body:\n{}'.format(resp.status_code, resp.content), resp)
mime_type = resp.headers['content-type']
if not any((typ in ['application/xml', 'text/xml'] for typ in mime_type.split(';'))):
raise HttpError(
'Metadata request did not return XML, MIME type: {}, body:\n{}'.format(mime_type, resp.content),
resp)
return resp.content
def _fetch_metadata(connection, url, logger):
# download metadata
logger.info('Fetching metadata')
resp = connection.get(url + '$metadata')
logger.debug('Retrieved the response:\n%s\n%s',
'\n'.join((f'H: {key}: {value}' for key, value in resp.headers.items())),
resp.content)
if resp.status_code != 200:
raise HttpError(
'Metadata request failed, status code: {}, body:\n{}'.format(resp.status_code, resp.content), resp)
mime_type = resp.headers['content-type']
if not any((typ in ['application/xml', 'text/xml'] for typ in mime_type.split(';'))):
raise HttpError(
'Metadata request did not return XML, MIME type: {}, body:\n{}'.format(mime_type, resp.content),
resp)
return resp.content
def __init__(self, message, response):
super(HttpError, self).__init__(message)
self.response = response
def get_entities_handler(response):
"""Gets entity set from HTTP Response"""
if response.status_code != requests.codes.ok:
raise HttpError('HTTP GET for Entity Set {0} failed with status code {1}'
.format(self._name, response.status_code), response)
content = response.json()
if isinstance(content, int):
return content
entities = content['d']['results']
result = []
for props in entities:
entity = EntityProxy(self._service, self._entity_set, self._entity_set.entity_type, props)
result.append(entity)
return result
def get_entity_handler(parent, nav_property, navigation_entity_set, response):
"""Gets entity from HTTP response"""
if response.status_code != requests.codes.ok:
raise HttpError('HTTP GET for Entity {0} failed with status code {1}'
.format(self._name, response.status_code), response)
entity = response.json()['d']
return NavEntityProxy(parent, nav_property, navigation_entity_set.entity_type, entity)
def delete_entity_handler(response):
"""Check if entity deletion was successful"""
if response.status_code != 204:
raise HttpError(f'HTTP POST for Entity delete {self._name} '
f'failed with status code {response.status_code}',
response)
def get_entity_handler(response):
"""Gets entity from HTTP response"""
if response.status_code != requests.codes.ok:
raise HttpError('HTTP GET for Entity {0} failed with status code {1}'
.format(self._name, response.status_code), response)
entity = response.json()['d']
return EntityProxy(self._service, self._entity_set, self._entity_set.entity_type, entity)
def update_entity_handler(response):
"""Gets modified entity encoded in HTTP Response"""
if response.status_code != 204:
raise HttpError('HTTP modify request for Entity Set {} failed with status code {}'
.format(self._name, response.status_code), response)