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_read_carto(mocker):
# Given
cm_mock = mocker.patch.object(ContextManager, 'copy_to')
cm_mock.return_value = GeoDataFrame({
'cartodb_id': [1, 2, 3],
'the_geom': [
'010100000000000000000000000000000000000000',
'010100000000000000000024400000000000002e40',
'010100000000000000000034400000000000003e40'
]
})
expected = GeoDataFrame({
'cartodb_id': [1, 2, 3],
'the_geom': [
Point([0, 0]),
Point([10, 15]),
Point([20, 30])
]
}, geometry='the_geom')
def test_rename_table(self, mocker):
# Given
def has_table(table_name):
if table_name == 'table_name':
return True
elif table_name == 'new_table_name':
return False
mocker.patch('cartoframes.core.managers.context_manager._create_auth_client')
mocker.patch.object(ContextManager, 'has_table', side_effect=has_table)
mock = mocker.patch.object(ContextManager, '_rename_table')
# When
cm = ContextManager(self.credentials)
result = cm.rename_table('table_name', 'NEW TABLE NAME')
# Then
mock.assert_called_once_with('table_name', 'new_table_name')
assert result == 'new_table_name'
def test_read_carto_retry_times(mocker):
# Given
mocker.patch.object(CartoDataFrame, 'set_geometry')
cm_mock = mocker.patch.object(ContextManager, 'copy_to')
# When
read_carto('__source__', CREDENTIALS, retry_times=1)
# Then
cm_mock.assert_called_once_with('__source__', None, None, 1)
def test_execute_query(self, mocker):
# Given
mocker.patch('cartoframes.core.managers.context_manager._create_auth_client')
mock = mocker.patch.object(SQLClient, 'send')
# When
cm = ContextManager(self.credentials)
cm.execute_query('query')
# Then
mock.assert_called_once_with('query', True, True, None)
def test_copy_from(self, mocker):
# Given
mocker.patch('cartoframes.core.managers.context_manager._create_auth_client')
mocker.patch.object(ContextManager, 'has_table', return_value=False)
mock = mocker.patch.object(ContextManager, '_copy_from')
cdf = CartoDataFrame({'A': [1]})
columns = [ColumnInfo('A', 'a', 'bigint', False)]
# When
cm = ContextManager(self.credentials)
cm.copy_from(cdf, 'TABLE NAME')
# Then
mock.assert_called_once_with(cdf, 'table_name', columns)
def test_rename_table_dest_exists_fail(self, mocker):
# Given
def has_table(table_name):
if table_name == 'table_name':
return True
elif table_name == 'new_table_name':
return True
mocker.patch('cartoframes.core.managers.context_manager._create_auth_client')
mocker.patch.object(ContextManager, 'has_table', side_effect=has_table)
# When
with pytest.raises(Exception) as e:
cm = ContextManager(self.credentials)
cm.rename_table('table_name', 'NEW TABLE NAME', 'fail')
# Then
assert str(e.value) == ('Table "new_table_name" already exists in your CARTO account. '
'Please choose a different `new_table_name` or use '
def setup_mocks(mocker, geom_type='point'):
mocker.patch.object(ContextManager, 'compute_query')
mocker.patch.object(ContextManager, 'get_geom_type', return_value=geom_type)
mocker.patch.object(ContextManager, 'get_bounds')
def __init__(self):
self._user_credentials = None
self._do_credentials = Credentials('do-metadata', 'default_public')
self._context_manager = ContextManager(self._do_credentials)
def has_table(table_name, credentials=None, schema=None):
"""
Check if the table exists in the CARTO account.
Args:
table_name (str): name of the table.
credentials (:py:class:`Credentials `, optional):
instance of Credentials (username, api_key, etc).
schema (str, optional):prefix of the table. By default, it gets the
`current_schema()` using the credentials.
"""
if not isinstance(table_name, str):
raise ValueError('Wrong table name. You should provide a valid table name.')
context_manager = ContextManager(credentials)
return context_manager.has_table(table_name, schema)
def __init__(self, source, credentials=None, geom_col=None):
self.credentials = None
if isinstance(source, str):
# Table, SQL query
self.type = SourceType.QUERY
self.manager = ContextManager(credentials)
self.query = self.manager.compute_query(source)
self.credentials = self.manager.credentials
elif isinstance(source, pandas.DataFrame):
# DataFrame, GeoDataFrame, CartoDataFrame
self.type = SourceType.GEOJSON
self.cdf = CartoDataFrame(source, copy=True)
if geom_col:
self.cdf.set_geometry(geom_col, inplace=True)
if not self.cdf.has_geometry():
raise Exception('No valid geometry found. Please provide an input source with ' +
'a valid geometry or specify the "geom_col" param with a geometry column.')
else:
raise ValueError('Wrong source input. Valid values are str and DataFrame.')