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_credentials_constructor(self):
credentials = Credentials(self.username, self.api_key)
assert credentials.api_key == self.api_key
assert credentials.username == self.username
assert credentials.base_url == self.base_url.strip('/')
def test_geography_download(self, mocked_bq_client, get_by_id_mock, get_all_mock):
# Given
get_by_id_mock.return_value = test_geography1
geography = Geography.get(test_geography1.id)
get_all_mock.return_value = [geography]
mocked_bq_client.return_value = BigQueryClientMock()
credentials = Credentials('fake_user', '1234')
# Then
geography.download('fake_path', credentials)
def test_dataset_download_raises_with_nonpurchased(self, mocked_bq_client, mocked_repo):
# mock dataset
mocked_repo.return_value = test_dataset1
# mock big query client
mocked_bq_client.return_value = BigQueryClientMock(NotFound('Fake error'))
# test
username = 'fake_user'
credentials = Credentials(username, '1234')
dataset = Dataset.get(test_dataset1.id)
with self.assertRaises(CartoException):
dataset.download(credentials)
def test_subscriptions_default_credentials(self, mocked_credentials, mocked_geographies, mocked_datasets):
# Given
expected_datasets = [test_dataset1, test_dataset2]
expected_geographies = [test_geography1, test_geography2]
expected_credentials = Credentials('user', '1234')
mocked_datasets.return_value = expected_datasets
mocked_geographies.return_value = expected_geographies
mocked_credentials.return_value = expected_credentials
catalog = Catalog()
# When
subscriptions = catalog.subscriptions()
# Then
mocked_datasets.assert_called_once_with({}, expected_credentials)
mocked_geographies.assert_called_once_with({}, expected_credentials)
assert isinstance(subscriptions, Subscriptions)
assert subscriptions.datasets == expected_datasets
assert subscriptions.geographies == expected_geographies
def test_get_all_geographies_credentials_without_do_enabled(self, mocked_repo):
# Given
def raise_exception(a, b):
raise ServerErrorException(['The user does not have Data Observatory enabled'])
mocked_repo.side_effect = raise_exception
credentials = Credentials('fake_user', '1234')
# When
with pytest.raises(Exception) as e:
Geography.get_all(credentials=credentials)
# Then
assert str(e.value) == (
'We are sorry, the Data Observatory is not enabled for your account yet. '
'Please contact your customer success manager or send an email to '
def test_source_get_credentials_username(self, mocker):
"""Source should return the correct credentials when username is provided"""
setup_mocks(mocker)
source = Source('faketable', credentials=Credentials(
username='fakeuser', api_key='1234'))
credentials = source.get_credentials()
assert credentials['username'] == 'fakeuser'
assert credentials['api_key'] == '1234'
assert credentials['base_url'] == 'https://fakeuser.carto.com'
def setup_method(self):
self.original_bigquery_Client = bigquery.Client
bigquery.Client = Mock(return_value=True)
self.original_storage_Client = storage.Client
storage.Client = Mock(return_value=True)
self.original_get_do_credentials = Credentials.get_do_credentials
Credentials.get_do_credentials = Mock(return_value=DoCredentials(_PUBLIC_PROJECT, _WORKING_PROJECT))
self.username = 'username'
self.apikey = 'apikey'
self.credentials = Credentials(self.username, self.apikey)
def test_geography_download_without_do_enabled(self, mocked_bq_client, get_by_id_mock, get_all_mock):
# Given
get_by_id_mock.return_value = test_geography1
geography = Geography.get(test_geography1.id)
get_all_mock.return_value = []
mocked_bq_client.return_value = BigQueryClientMock(
ServerErrorException(['The user does not have Data Observatory enabled'])
)
credentials = Credentials('fake_user', '1234')
# When
with pytest.raises(Exception) as e:
geography.download('fake_path', credentials)
# Then
assert str(e.value) == (
'We are sorry, the Data Observatory is not enabled for your account yet. '
'Please contact your customer success manager or send an email to '
def test_credentials_baseurl_without_https(self):
with pytest.raises(ValueError):
Credentials(api_key=self.api_key, base_url=self.base_url.replace('https', 'http'))
def test_prepare_variables_without_agg_method(self, get_mock, _validate_bq_operations_mock):
_validate_bq_operations_mock.return_value = True
variable_id = 'project.dataset.table.variable'
variable = Variable({
'id': variable_id,
'column_name': 'column',
'dataset_id': 'fake_name',
'agg_method': None
})
get_mock.return_value = variable
credentials = Credentials('fake_user', '1234')
one_variable_cases = [
variable_id,
variable
]
for case in one_variable_cases:
result = prepare_variables(case, credentials)
assert result == [variable]
for case in one_variable_cases:
result = prepare_variables(case, credentials, aggregation={})
assert result == []