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_upload_pandas_categorical(self, con):
con.execute("DROP TABLE IF EXISTS test_categorical;")
df = pd.DataFrame({"A": ["a", "b", "c", "a"]})
df["B"] = df["A"].astype('category')
# test that table created correctly when it doesn't exist on server
con.load_table("test_categorical", df)
ans = con.execute("select * from test_categorical").fetchall()
assert ans == [('a', 'a'), ('b', 'b'), ('c', 'c'), ('a', 'a')]
assert con.get_table_details("test_categorical") == \
[ColumnDetails(name='A', type='STR', nullable=True, precision=0,
scale=0, comp_param=32, encoding='DICT',
is_array=False),
ColumnDetails(name='B', type='STR', nullable=True, precision=0,
scale=0, comp_param=32, encoding='DICT',
is_array=False)]
# load row-wise
con.load_table("test_categorical", df, method="rows")
# load columnar
con.load_table("test_categorical", df, method="columnar")
# load arrow
con.load_table("test_categorical", df, method="arrow")
# test end result
scale=0, comp_param=0),
is_reserved_keyword=False, src_name='')]
result = _extract_column_details(data)
expected = [
ColumnDetails(name='date_', type='STR', nullable=True, precision=0,
scale=0, comp_param=32, encoding='DICT',
is_array=False),
ColumnDetails(name='trans', type='STR', nullable=True, precision=0,
scale=0, comp_param=32, encoding='DICT',
is_array=False),
ColumnDetails(name='symbol', type='STR', nullable=True,
precision=0, scale=0, comp_param=32,
encoding='DICT',
is_array=False),
ColumnDetails(name='qty', type='INT', nullable=True, precision=0,
scale=0, comp_param=0, encoding='NONE',
is_array=False),
ColumnDetails(name='price', type='FLOAT', nullable=True,
precision=0, scale=0, comp_param=0, encoding='NONE',
is_array=False),
ColumnDetails(name='vol', type='FLOAT', nullable=True, precision=0,
scale=0, comp_param=0, encoding='NONE',
is_array=False)
]
assert result == expected
ColumnDetails(name='trans', type='STR', nullable=True, precision=0,
scale=0, comp_param=32, encoding='DICT',
is_array=False),
ColumnDetails(name='symbol', type='STR', nullable=True,
precision=0, scale=0, comp_param=32,
encoding='DICT', is_array=False),
ColumnDetails(name='qty', type='INT', nullable=True, precision=0,
scale=0, comp_param=0, encoding='NONE',
is_array=False),
ColumnDetails(name='price', type='FLOAT', nullable=True,
precision=0, scale=0, comp_param=0, encoding='NONE',
is_array=False),
ColumnDetails(name='vol', type='FLOAT', nullable=True, precision=0,
scale=0, comp_param=0, encoding='NONE',
is_array=False),
ColumnDetails(name='exchanges', type='STR', nullable=True,
precision=0, scale=0, comp_param=32, encoding='DICT',
is_array=True)
]
assert result == expected
c.execute('drop table if exists stocks;')