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_limit(mocker):
# Given
mocker.patch.object(CartoDataFrame, 'set_geometry')
cm_mock = mocker.patch.object(ContextManager, 'copy_to')
# When
read_carto('__source__', CREDENTIALS, limit=1)
# Then
cm_mock.assert_called_once_with('__source__', None, 1, 3)
def test_merge(self):
cdf1 = CartoDataFrame({'lkey': ['foo', 'bar'], 'value': [1, 2]})
cdf2 = CartoDataFrame({'rkey': ['foo', 'bar'], 'value': [5, 6]})
cdf = cdf1.merge(cdf2, left_on='lkey', right_on='rkey')
assert isinstance(cdf, CartoDataFrame)
def test_geocode_dataframe_with_custom_status(self):
self.skip(if_no_credits=True, if_no_credentials=True)
gc = Geocoding(credentials=self.credentials)
cdf = CartoDataFrame([['Gran Via 46', 'Madrid'], ['Ebro 1', 'Sevilla']], columns=['address', 'city'])
quota = self.used_quota(gc)
status = {'gc_rel': 'relevance'}
# Preview
info = gc.geocode(cdf, street='address', city='city', country={'value': 'Spain'},
status=status, dry_run=True).metadata
self.assertEqual(info.get('required_quota'), 2)
self.assertEqual(self.used_quota(gc), quota)
# Geocode
gc_df, info = gc.geocode(cdf, street='address', city='city', country={'value': 'Spain'}, status=status)
self.assertTrue(isinstance(gc_df, pd.DataFrame))
self.assertEqual(info.get('required_quota'), 2)
self.assertEqual(info.get('successfully_geocoded'), 2)
def test_dissolve(self):
cdf = CartoDataFrame({'a': [1], 'geometry': [Point(0, 0)]})
cdf = cdf.dissolve(by='a')
assert isinstance(cdf, CartoDataFrame)
def test_prepare_data(self):
geom_column = 'the_geom'
enrichment_service = EnrichmentService(credentials=self.credentials)
point = Point(1, 1)
df = pd.DataFrame(
[[1, point]],
columns=['cartodb_id', geom_column])
expected_cdf = CartoDataFrame(
[[1, point, 0, to_geojson(point)]],
columns=['cartodb_id', geom_column, _ENRICHMENT_ID, _GEOM_COLUMN],
geometry=geom_column
)
result = enrichment_service._prepare_data(df, geom_column)
assert result.equals(expected_cdf)
def test_upload_data(self):
geom_column = 'the_geom'
user_dataset = 'test_dataset'
point = Point(1, 1)
input_cdf = CartoDataFrame(
[[1, point, 0, to_geojson(point)]],
columns=['cartodb_id', geom_column, _ENRICHMENT_ID, _GEOM_COLUMN],
geometry=geom_column
)
expected_schema = {_ENRICHMENT_ID: 'INTEGER', _GEOM_COLUMN: 'GEOGRAPHY'}
expected_cdf = CartoDataFrame(
[[0, to_geojson(point)]],
columns=[_ENRICHMENT_ID, _GEOM_COLUMN])
# mock
def assert_upload_data(_, dataframe, schema, tablename):
assert dataframe.equals(expected_cdf)
assert schema == expected_schema
assert isinstance(tablename, str) and len(tablename) > 0
assert tablename == user_dataset
enrichment_service = EnrichmentService(credentials=self.credentials)
original = BigQueryClient.upload_dataframe
BigQueryClient.upload_dataframe = assert_upload_data
enrichment_service._upload_data(user_dataset, input_cdf)
BigQueryClient.upload_dataframe = original
def test_invalid_arguments(self):
gc = Geocoding(credentials=self.credentials)
cdf = CartoDataFrame([['Gran Via 46', 'Madrid'], ['Ebro 1', 'Sevilla']], columns=['address', 'city'])
with self.assertRaises(ValueError):
gc.geocode(cdf, street='address', city={'columna': 'city'})
with self.assertRaises(ValueError):
gc.geocode(cdf, street='address', state={'columna': 'city'})
with self.assertRaises(ValueError):
gc.geocode(cdf, street='address', country={'columna': 'city'})
with self.assertRaises(ValueError):
gc.geocode(cdf, street='address', city={'column': 'ciudad'})
with self.assertRaises(ValueError):
gc.geocode(cdf, street='address', state={'column': 'ciudad'})
with self.assertRaises(ValueError):
gc.geocode(cdf, street='address', country={'column': 'ciudad'})
with self.assertRaises(ValueError):
gc.geocode(cdf, street='address', city='ciudad')
with self.assertRaises(ValueError):
gc.geocode(cdf, street='address', state='ciudad')
def test_plot_series(self, mocker):
series_plot_mock = mocker.patch.object(Series, 'plot')
cdf = CartoDataFrame({'x': [1]})
cdf['x'].plot()
series_plot_mock.assert_called_once_with()
def test_explode(self):
cdf = CartoDataFrame({'a': [1], 'geometry': [Point(0, 0)]})
cdf = cdf.explode()
assert isinstance(cdf, CartoDataFrame)
def test_constructor(self):
cdf = CartoDataFrame(self.gdf)
assert isinstance(cdf, CartoDataFrame)
assert cdf._constructor == CartoDataFrame