<s> import StringIO <EOL> import apsw <EOL> import contextlib <EOL> import pytest <EOL> import tempfile <EOL> import bayeslite <EOL> import bayeslite . guess <EOL> from bayeslite import bql_quote_name <EOL> import test_core <EOL> @ contextlib . contextmanager <EOL> def bayesdb_csv_stream ( csv ) : <EOL> with test_core . bayesdb ( ) as bdb : <EOL> yield ( bdb , StringIO . StringIO ( csv ) ) <EOL> @ contextlib . contextmanager <EOL> def bayesdb_csv_file ( csv ) : <EOL> with test_core . bayesdb ( ) as bdb : <EOL> with tempfile . NamedTemporaryFile ( prefix = '<STR_LIT>' ) as f : <EOL> with open ( f . name , '<STR_LIT:w>' ) as out : <EOL> out . write ( csv ) <EOL> yield ( bdb , f . name ) <EOL> def test_csv_import_empty ( ) : <EOL> with bayesdb_csv_stream ( '<STR_LIT>' ) as ( bdb , f ) : <EOL> with pytest . raises ( IOError ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , <EOL> create = True ) <EOL> def test_csv_import_nocols ( ) : <EOL> with bayesdb_csv_stream ( '<STR_LIT:\n>' ) as ( bdb , f ) : <EOL> with pytest . raises ( IOError ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , <EOL> create = True ) <EOL> def test_csv_import_onecol_key ( ) : <EOL> with bayesdb_csv_stream ( '<STR_LIT>' ) as ( bdb , f ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , <EOL> create = True ) <EOL> with pytest . raises ( ValueError ) : <EOL> bayeslite . guess . bayesdb_guess_generator ( bdb , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_csv_import_onecol ( ) : <EOL> with bayesdb_csv_stream ( '<STR_LIT>' ) as ( bdb , f ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , create = True ) <EOL> def test_csv_import_toofewcols ( ) : <EOL> with bayesdb_csv_stream ( '<STR_LIT>' ) as ( bdb , f ) : <EOL> with pytest . raises ( IOError ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , create = True ) <EOL> def test_csv_import_toomanycols ( ) : <EOL> with bayesdb_csv_stream ( '<STR_LIT>' ) as ( bdb , f ) : <EOL> with pytest . raises ( IOError ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , create = True ) <EOL> def test_csv_import_dupcols ( ) : <EOL> with bayesdb_csv_stream ( '<STR_LIT>' ) as ( bdb , f ) : <EOL> with pytest . raises ( IOError ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , create = True ) <EOL> with bayesdb_csv_stream ( '<STR_LIT>' ) as ( bdb , f ) : <EOL> with pytest . raises ( IOError ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , create = True ) <EOL> csv_data = '''<STR_LIT>''' <EOL> def test_csv_import ( ) : <EOL> with bayesdb_csv_stream ( csv_data ) as ( bdb , f ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , <EOL> create = True ) <EOL> def test_csv_import_schema ( ) : <EOL> with bayesdb_csv_stream ( csv_data ) as ( bdb , f ) : <EOL> bdb . sql_execute ( '''<STR_LIT>''' ) <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , <EOL> create = False ) <EOL> bdb . execute ( '<STR_LIT>' ) . fetchall ( ) <EOL> with pytest . raises ( apsw . SQLError ) : <EOL> bdb . execute ( '<STR_LIT>' ) <EOL> raise apsw . SQLError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> bdb . execute ( '''<STR_LIT>''' ) <EOL> bdb . execute ( '<STR_LIT>' ) . fetchall ( ) <EOL> with pytest . raises ( bayeslite . BQLError ) : <EOL> bdb . execute ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_csv_import_schema_case ( ) : <EOL> with bayesdb_csv_stream ( csv_data ) as ( bdb , f ) : <EOL> bdb . sql_execute ( '''<STR_LIT>''' ) <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , <EOL> create = False ) <EOL> bayeslite . guess . bayesdb_guess_generator ( bdb , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_csv_import_badschema0 ( ) : <EOL> with bayesdb_csv_stream ( csv_data ) as ( bdb , f ) : <EOL> bdb . sql_execute ( '''<STR_LIT>''' ) <EOL> with pytest . raises ( IOError ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , <EOL> create = False ) <EOL> def test_csv_import_badschema1 ( ) : <EOL> with bayesdb_csv_stream ( csv_data ) as ( bdb , f ) : <EOL> bdb . sql_execute ( '''<STR_LIT>''' ) <EOL> with pytest . raises ( IOError ) : <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT>' , f , header = True , <EOL> create = False ) <EOL> csv_data_missing = '''<STR_LIT>''' <EOL> def test_csv_missing ( ) : <EOL> with bayesdb_csv_stream ( csv_data_missing ) as ( bdb , f ) : <EOL> bdb . sql_execute ( '<STR_LIT>' ) <EOL> bayeslite . bayesdb_read_csv ( bdb , '<STR_LIT:t>' , f , header = True , create = False ) <EOL> def clean ( column_name ) : <EOL> qcn = bql_quote_name ( column_name ) <EOL> sql = "<STR_LIT>" % ( qcn , qcn , qcn ) <EOL> bdb . sql_execute ( sql ) <EOL> clean ( '<STR_LIT:a>' ) <EOL> clean ( '<STR_LIT:b>' ) <EOL> clean ( '<STR_LIT:c>' ) <EOL> assert bdb . execute ( '<STR_LIT>' ) . fetchall ( ) == [ <EOL> ( <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ) , <EOL> ( <NUM_LIT> , None , <NUM_LIT> ) , <EOL> ( <NUM_LIT> , <NUM_LIT> , None ) , <EOL> ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) , <EOL> ] </s>
