<s> import unittest <EOL> import threading <EOL> import time <EOL> from brewery import ds <EOL> import brewery . metadata <EOL> from sqlalchemy import Table , Column , Integer , String , Text <EOL> from sqlalchemy import create_engine , MetaData <EOL> class SQLStreamsTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . engine = create_engine ( "<STR_LIT>" ) <EOL> self . metadata = MetaData ( ) <EOL> self . fields = brewery . metadata . FieldList ( [ <EOL> ( "<STR_LIT>" , "<STR_LIT:string>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:string>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:string>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:string>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:string>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) ] ) <EOL> self . example_row = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:foo>" , <NUM_LIT> , <NUM_LIT:100> ] <EOL> def test_table_fields ( self ) : <EOL> table = Table ( '<STR_LIT>' , self . metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:32> ) ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:255> ) ) , <EOL> Column ( '<STR_LIT>' , Text ) <EOL> ) <EOL> self . metadata . create_all ( self . engine ) <EOL> stream = ds . SQLDataSource ( connection = self . engine , table = str ( table ) ) <EOL> fields = stream . fields <EOL> self . assertEqual ( <NUM_LIT:4> , len ( fields ) ) <EOL> def test_target_no_existing_table ( self ) : <EOL> stream = ds . SQLDataTarget ( connection = self . engine , table = "<STR_LIT:test>" ) <EOL> self . assertRaises ( Exception , stream . initialize ) <EOL> def test_target_create_table ( self ) : <EOL> stream = ds . SQLDataTarget ( connection = self . engine , table = "<STR_LIT:test>" , create = True ) <EOL> self . assertRaises ( Exception , stream . initialize ) <EOL> stream . fields = self . fields <EOL> stream . initialize ( ) <EOL> cnames = [ str ( c ) for c in stream . table . columns ] <EOL> fnames = [ "<STR_LIT>" + f . name for f in self . fields ] <EOL> self . assertEqual ( fnames , cnames ) <EOL> stream . finalize ( ) <EOL> def test_target_replace_table ( self ) : <EOL> table = Table ( '<STR_LIT:test>' , self . metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:32> ) ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:255> ) ) , <EOL> Column ( '<STR_LIT>' , Text ) <EOL> ) <EOL> self . metadata . create_all ( self . engine ) <EOL> stream = ds . SQLDataTarget ( connection = self . engine , table = "<STR_LIT:test>" , <EOL> create = True , replace = False ) <EOL> stream . fields = self . fields <EOL> self . assertRaises ( Exception , stream . initialize ) <EOL> stream = ds . SQLDataTarget ( connection = self . engine , table = "<STR_LIT:test>" , <EOL> create = True , replace = True ) <EOL> stream . fields = self . fields <EOL> stream . initialize ( ) <EOL> cnames = [ str ( c ) for c in stream . table . columns ] <EOL> fnames = [ "<STR_LIT>" + f . name for f in self . fields ] <EOL> self . assertEqual ( fnames , cnames ) <EOL> stream . finalize ( ) <EOL> def test_target_concrete_type_map ( self ) : <EOL> ctm = { "<STR_LIT:string>" : String ( <NUM_LIT> ) } <EOL> stream = ds . SQLDataTarget ( connection = self . engine , table = "<STR_LIT:test>" , <EOL> create = True , <EOL> fields = self . fields , <EOL> concrete_type_map = ctm ) <EOL> stream . initialize ( ) <EOL> c = stream . table . c [ "<STR_LIT>" ] <EOL> self . assertEqual ( <NUM_LIT> , c . type . length ) </s>
