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_cartocontext_fetch_empty(self):
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey)
# empty response
df_empty = cc.fetch('''
SELECT 1
LIMIT 0
''')
# no rows, one column
self.assertTupleEqual(df_empty.shape, (0, 1))
# is a DataFrame
self.assertIsInstance(df_empty, pd.DataFrame)
def test_cartocontext_execute(self):
"""context.CartoContext.execute"""
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey)
df = pd.DataFrame({'vals': list('abcd'), 'ids': list('wxyz')})
df = df.astype({'vals': str, 'ids': str})
cc.write(df, self.test_write_table, overwrite=True)
self.assertEquals(Dataset(cc, self.test_write_table).exists(), True)
cc.execute('''
DROP TABLE {table_name}
'''.format(table_name=self.test_write_table))
self.assertEquals(Dataset(cc, self.test_write_table).exists(), False)
def test_batchjobstatus_methods(self):
"""context.BatchJobStatus methods"""
from cartoframes.batch import BatchJobStatus
from carto.sql import BatchSQLClient
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey)
batch_client = BatchSQLClient(cc.auth_client)
job_response = batch_client.create(['select 1', ])
job_status = BatchJobStatus(cc, job_response)
possible_status = ('pending', 'running', 'done',
'canceled', 'unknown', )
self.assertTrue(job_status.get_status() in possible_status)
job_status._set_status('foo')
self.assertEqual(job_status.get_status(), 'foo')
new_status = job_status.status()
self.assertSetEqual(set(new_status.keys()),
{'status', 'updated_at', 'created_at'})
def test_batchjobstatus_repr(self):
"""context.BatchJobStatus.__repr__"""
from cartoframes.batch import BatchJobStatus
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey)
bjs = BatchJobStatus(cc, dict(job_id='foo', status='unknown',
created_at=None))
self.assertMultiLineEqual(bjs.__repr__(),
("BatchJobStatus(job_id='foo', "
"last_status='unknown', "
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey,
verbose=True)
# request-response usage
resp = requests.get('http://httpbin.org/get')
cc._debug_print(resp=resp)
cc._debug_print(resp=resp.text)
# non-requests-response usage
test_str = 'this is a test'
long_test_str = ', '.join([test_str] * 100)
self.assertIsNone(cc._debug_print(test_str=test_str))
self.assertIsNone(cc._debug_print(long_str=long_test_str))
# verbose = False test
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey,
verbose=False)
self.assertIsNone(cc._debug_print(resp=test_str))
def test_cartocontext_write_index(self):
"""context.CartoContext.write with non-default index"""
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey)
df = pd.DataFrame({'vals': range(3), 'ids': list('abc')},
index=list('xyz'))
df.index.name = 'named_index'
dataset = cc.write(df, self.write_named_index)
self.write_named_index = dataset.table_name
df_index = cc.read(self.write_named_index)
self.assertSetEqual(set(('the_geom', 'vals', 'ids', 'named_index')),
set(df_index.columns))
def tearDown(self):
"""restore to original state"""
tables = (self.test_write_table,
self.test_write_batch_table,
self.test_write_lnglat_table,
self.test_query_table,
self.mixed_case_table.lower(),
self.write_named_index, )
sql_drop = 'DROP TABLE IF EXISTS {};'
if self.apikey and self.baseurl:
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey)
for table in tables:
try:
cc.delete(table)
cc.sql_client.send(sql_drop.format(table))
except CartoException:
warnings.warn('Error deleting tables')
def test_cartocontext_fetch_with_exception(self):
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey)
# see what happens if a query fails after 100 successful rows
with self.assertRaises(CartoException):
cc.fetch('''
WITH cte AS (
def test_cartocontext_query(self):
"""context.CartoContext.query"""
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey)
cols = ('link', 'body', 'displayname', 'friendscount', 'postedtime', )
df = cc.query('''
SELECT {cols}, '02-06-1429'::date as invalid_df_date
FROM tweets_obama
LIMIT 100
'''.format(cols=','.join(cols)))
# ensure columns are in expected order
df = df[list(cols) + ['invalid_df_date', ]]
# same number of rows
self.assertEqual(len(df), 100,
msg='Expected number or rows')
# same type of object
def test_cartocontext_map_geom_type(self):
"""context.CartoContext.map basemap geometry type defaults"""
from cartoframes import Layer, QueryLayer
cc = cartoframes.CartoContext(base_url=self.baseurl,
api_key=self.apikey)
# baseid1 = dark, labels1 = labels on top in named map name
labels_polygon = cc.map(layers=Layer(self.test_read_table))
self.assertRegexpMatches(labels_polygon.__html__(),
'.*baseid2_labels1.*',
msg='labels should be on top since only a '
'polygon layer is present')
# baseid2 = voyager, labels0 = labels on bottom
labels_point = cc.map(layers=Layer(self.test_point_table))
self.assertRegexpMatches(labels_point.__html__(),
'.*baseid2_labels0.*',
msg='labels should be on bottom because a '
'point layer is present')