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_valid_operation(self):
"""cluster_size_layer should raise an error if the operation is invalid"""
msg = '"invalid" is not a valid operation. Valid operations are count, avg, min, max, sum'
with pytest.raises(CartoException) as e:
helpers.cluster_size_layer(
source=self.source,
value='name',
operation='invalid'
)
assert str(e.value) == msg
def test_default_legend(self):
"""Map should raise an error if default_legend is True but there is no title"""
msg = 'The default legend needs a map title to be displayed'
with pytest.raises(CartoException) as e:
Map(default_legend=True)
assert str(e.value) == msg
def test__init_maps_valid(self):
"""Layout should raise an error if any element in the map list is not a Map"""
msg = 'All the elements in the Layout should be an instance of Map'
with pytest.raises(CartoException) as e:
Layout([Layer(Source(SOURCE))])
assert str(e.value) == msg
def test_map_publish_unsync_fails(self):
query = "SELECT 1"
dataset = DatasetMock.from_query(query=query, context=self.context)
dataset._is_saved_in_carto = False
map = MapMock(Layer(Source(dataset)))
msg = 'The map layers are not synchronized with CARTO. Please, use the `sync_data` before publishing the map'
with self.assertRaises(CartoException, msg=msg):
map.publish('test', context=self.context)
response = self.client.send(url,
http_method=http_method,
params=params,
stream=True)
response.raise_for_status()
except CartoRateLimitException as e:
raise e
except HTTPError as e:
if 400 <= response.status_code < 500:
# Client error, provide better reason
reason = response.json()['error'][0]
error_msg = u'%s Client Error: %s' % (response.status_code,
reason)
raise CartoException(error_msg)
else:
raise CartoException(e)
except Exception as e:
raise CartoException(e)
return response
def delete(self):
if self.dependent_visualizations_count > 0:
raise CartoException(_(
'This dataset contains dependent visualizations. ' +
'Delete them to be able to delete this dataset or use `force_delete` ' +
'to delete the dataset and the dependent visualizations.')
)
super(WarnResource, self).delete()
:param client_params: To be send to the CARTO API. See CARTO's
documentation depending on the subclass
you are using
:type client_params: kwargs
:return:
:raise: CartoException
"""
try:
self.send(self.get_collection_endpoint(),
http_method="POST",
**client_params)
except Exception as e:
raise CartoException(e)
def download_to_file(self, job, file_path, fail_if_exists=False, column_names=None, progress_bar=True):
if fail_if_exists and os.path.isfile(file_path):
raise CartoException('The file `{}` already exists.'.format(file_path))
try:
rows = self._download_by_bq_storage_api(job)
except Exception:
log.debug('Cannot download using BigQuery Storage API, fallback to standard')
try:
rows = job.result()
except Exception:
if job.errors:
log.error([error['message'] for error in job.errors if 'message' in error])
raise CartoException('Error downloading data')
_rows_to_file(rows, file_path, column_names, progress_bar)
def _send_map_template(self, layers, has_zoom):
map_name = get_map_name(layers, has_zoom=has_zoom)
if map_name not in self._map_templates:
resp = self._auth_send(
'api/v1/map/named', 'POST',
headers={'Content-Type': 'application/json'},
data=get_map_template(layers, has_zoom=has_zoom))
if 'errors' in resp:
resp = self._auth_send(
'api/v1/map/named/{}'.format(map_name),
'PUT',
headers={'Content-Type': 'application/json'},
data=get_map_template(layers, has_zoom=has_zoom))
if 'errors' in resp:
raise CartoException(resp)
self._map_templates[map_name] = True
return map_name
def filter(self, **search_args):
"""
Should get all the current users from CARTO, but this is currently not
supported by the API
"""
raise CartoException(_("Retrieving user list is not currently \
supported by the API"))