Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
content_type = response.headers.get('Content-Type')
error_message = ''
if content_type is None or content_type.find('text/plain') < 0:
try:
wrapped_exception = self._base_deserialize('WrappedException', response)
if wrapped_exception is not None and wrapped_exception.message is not None:
raise AzureDevOpsServiceError(wrapped_exception)
else:
# System exceptions from controllers are not returning wrapped exceptions.
# Following code is to handle this unusual exception json case.
# TODO: dig into this.
collection_wrapper = self._base_deserialize('VssJsonCollectionWrapper', response)
if collection_wrapper is not None and collection_wrapper.value is not None:
wrapped_exception = self._base_deserialize('ImproperException', collection_wrapper.value)
if wrapped_exception is not None and wrapped_exception.message is not None:
raise AzureDevOpsClientRequestError(wrapped_exception.message)
# if we get here we still have not raised an exception, try to deserialize as a System Exception
system_exception = self._base_deserialize('SystemException', response)
if system_exception is not None and system_exception.message is not None:
raise AzureDevOpsClientRequestError(system_exception.message)
except DeserializationError:
pass
elif response.content is not None:
error_message = response.content.decode("utf-8") + ' '
if response.status_code == 401:
full_message_format = '{error_message}The requested resource requires user authentication: {url}'
raise AzureDevOpsAuthenticationError(full_message_format.format(error_message=error_message,
url=request.url))
else:
full_message_format = '{error_message}Operation returned a {status_code} status code.'
raise AzureDevOpsClientRequestError(full_message_format.format(error_message=error_message,
status_code=response.status_code))
# Last resort, make the call to the server
options_uri = self._combine_url(self.config.base_url, '_apis')
request = ClientRequest(method='OPTIONS', url=self._client.format_url(options_uri))
if all_host_types:
query_parameters = {'allHostTypes': True}
request.format_parameters(query_parameters)
headers = {'Accept': 'application/json'}
if self._suppress_fedauth_redirect:
headers['X-TFS-FedAuthRedirect'] = 'Suppress'
if self._force_msa_pass_through:
headers['X-VSS-ForceMsaPassThrough'] = 'true'
response = self._send_request(request, headers=headers)
wrapper = self._base_deserialize('VssJsonCollectionWrapper', response)
if wrapper is None:
raise AzureDevOpsClientRequestError("Failed to retrieve resource locations from: {}".format(options_uri))
collection = wrapper.value
returned_locations = self._base_deserialize('[ApiResourceLocation]',
collection)
if all_host_types:
self._all_host_types_locations = returned_locations
else:
self._locations = returned_locations
try:
OPTIONS_FILE_CACHE[self.normalized_url] = wrapper.value
except SerializationError as ex:
logger.debug(ex, exc_info=True)
return returned_locations
if resource_areas is None:
raise AzureDevOpsClientRequestError(('Failed to retrieve resource areas '
+ 'from server: {url}').format(url=self.base_url))
if not resource_areas:
# For OnPrem environments we get an empty list.
return self.base_url
for resource_area in resource_areas:
if resource_area.id.lower() == resource_id.lower():
return resource_area.location_url
# Check SPS deployment level for the resource area
resource_area = self._get_deployment_resource_area_from_sps(resource_id)
if resource_area is not None:
return resource_area.location_url
raise AzureDevOpsClientRequestError(('Could not find information for resource area {id} '
+ 'from server: {url}').format(id=resource_id,
url=self.base_url))
def _get_url_for_client_instance(self, client_class):
resource_id = client_class.resource_area_identifier
if resource_id is None:
return self.base_url
else:
resource_areas = self._get_resource_areas()
if resource_areas is None:
raise AzureDevOpsClientRequestError(('Failed to retrieve resource areas '
+ 'from server: {url}').format(url=self.base_url))
if not resource_areas:
# For OnPrem environments we get an empty list.
return self.base_url
for resource_area in resource_areas:
if resource_area.id.lower() == resource_id.lower():
return resource_area.location_url
# Check SPS deployment level for the resource area
resource_area = self._get_deployment_resource_area_from_sps(resource_id)
if resource_area is not None:
return resource_area.location_url
raise AzureDevOpsClientRequestError(('Could not find information for resource area {id} '
+ 'from server: {url}').format(id=resource_id,
url=self.base_url))
wrapped_exception = self._base_deserialize('WrappedException', response)
if wrapped_exception is not None and wrapped_exception.message is not None:
raise AzureDevOpsServiceError(wrapped_exception)
else:
# System exceptions from controllers are not returning wrapped exceptions.
# Following code is to handle this unusual exception json case.
# TODO: dig into this.
collection_wrapper = self._base_deserialize('VssJsonCollectionWrapper', response)
if collection_wrapper is not None and collection_wrapper.value is not None:
wrapped_exception = self._base_deserialize('ImproperException', collection_wrapper.value)
if wrapped_exception is not None and wrapped_exception.message is not None:
raise AzureDevOpsClientRequestError(wrapped_exception.message)
# if we get here we still have not raised an exception, try to deserialize as a System Exception
system_exception = self._base_deserialize('SystemException', response)
if system_exception is not None and system_exception.message is not None:
raise AzureDevOpsClientRequestError(system_exception.message)
except DeserializationError:
pass
elif response.content is not None:
error_message = response.content.decode("utf-8") + ' '
if response.status_code == 401:
full_message_format = '{error_message}The requested resource requires user authentication: {url}'
raise AzureDevOpsAuthenticationError(full_message_format.format(error_message=error_message,
url=request.url))
else:
full_message_format = '{error_message}Operation returned a {status_code} status code.'
raise AzureDevOpsClientRequestError(full_message_format.format(error_message=error_message,
status_code=response.status_code))
raise AzureDevOpsClientRequestError(wrapped_exception.message)
# if we get here we still have not raised an exception, try to deserialize as a System Exception
system_exception = self._base_deserialize('SystemException', response)
if system_exception is not None and system_exception.message is not None:
raise AzureDevOpsClientRequestError(system_exception.message)
except DeserializationError:
pass
elif response.content is not None:
error_message = response.content.decode("utf-8") + ' '
if response.status_code == 401:
full_message_format = '{error_message}The requested resource requires user authentication: {url}'
raise AzureDevOpsAuthenticationError(full_message_format.format(error_message=error_message,
url=request.url))
else:
full_message_format = '{error_message}Operation returned a {status_code} status code.'
raise AzureDevOpsClientRequestError(full_message_format.format(error_message=error_message,
status_code=response.status_code))