<s> from com . ziclix . python . sql import zxJDBC <EOL> from java . util import Calendar , Date as JDate <EOL> import tempfile , os , time , runner <EOL> class zxCoreTestCase ( runner . SQLTestCase ) : <EOL> def setUp ( self ) : <EOL> runner . SQLTestCase . setUp ( self ) <EOL> self . db = self . connect ( ) <EOL> self . db . autocommit = <NUM_LIT:0> <EOL> def tearDown ( self ) : <EOL> self . db . close ( ) <EOL> runner . SQLTestCase . tearDown ( self ) <EOL> def connect ( self ) : <EOL> factory = runner . __imp__ ( self . factory . classname ) <EOL> args = map ( lambda x : x [ <NUM_LIT:1> ] , self . factory . arguments ) <EOL> connect = getattr ( factory , self . factory . method ) <EOL> return apply ( connect , args , self . factory . keywords ) <EOL> def cursor ( self , * args , ** kws ) : <EOL> c = apply ( self . db . cursor , args , kws ) <EOL> if hasattr ( self , "<STR_LIT>" ) : <EOL> c . datahandler = self . datahandler ( c . datahandler ) <EOL> return c <EOL> class zxJDBCTestCase ( zxCoreTestCase ) : <EOL> def setUp ( self ) : <EOL> zxCoreTestCase . setUp ( self ) <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . db . commit ( ) <EOL> except : <EOL> self . db . rollback ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . db . commit ( ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . db . commit ( ) <EOL> finally : <EOL> c . close ( ) <EOL> def tearDown ( self ) : <EOL> c = self . cursor ( ) <EOL> try : <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> except : <EOL> self . db . rollback ( ) <EOL> finally : <EOL> c . close ( ) <EOL> zxCoreTestCase . tearDown ( self ) <EOL> class zxAPITestCase ( zxJDBCTestCase ) : <EOL> def testConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . db , "<STR_LIT>" <EOL> def testAutocommit ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . db . __connection__ . getMetaData ( ) . supportsTransactions ( ) : <EOL> self . db . autocommit = <NUM_LIT:1> <EOL> self . assertEquals ( <NUM_LIT:1> , self . db . __connection__ . getAutoCommit ( ) ) <EOL> self . db . autocommit = <NUM_LIT:0> <EOL> self . assertEquals ( <NUM_LIT:0> , self . db . __connection__ . getAutoCommit ( ) ) <EOL> def testSimpleQuery ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> f = c . fetchall ( ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" <EOL> c . execute ( "<STR_LIT>" ) <EOL> data = c . fetchone ( ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" <EOL> assert data [ <NUM_LIT:0> ] == <NUM_LIT:1> , "<STR_LIT>" % ( data [ <NUM_LIT:0> ] ) <EOL> finally : <EOL> c . close ( ) <EOL> def testNoneQuery ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( None ) <EOL> finally : <EOL> c . close ( ) <EOL> def _test_preparedstatement ( self , dynamic ) : <EOL> c = self . cursor ( dynamic ) <EOL> try : <EOL> p = c . prepare ( "<STR_LIT>" ) <EOL> for i in range ( <NUM_LIT:1> , <NUM_LIT:8> ) : <EOL> c . execute ( p , ( i , ) ) <EOL> data = c . fetchall ( ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( data ) ) <EOL> assert not p . closed <EOL> p . close ( ) <EOL> assert p . closed <EOL> self . assertRaises ( zxJDBC . ProgrammingError , c . execute , p , ( <NUM_LIT:1> , ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testStaticPrepare ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_preparedstatement ( <NUM_LIT:0> ) <EOL> def testDynamicPrepare ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_preparedstatement ( <NUM_LIT:1> ) <EOL> def _test_cursorkeywords ( self , * args , ** kws ) : <EOL> c = self . cursor ( * args , ** kws ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> data = c . fetchmany ( <NUM_LIT:1> ) <EOL> assert len ( data ) == <NUM_LIT:1> , "<STR_LIT>" <EOL> finally : <EOL> c . close ( ) <EOL> def testCursorKeywords ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_cursorkeywords ( dynamic = <NUM_LIT:1> ) <EOL> self . _test_cursorkeywords ( dynamic = <NUM_LIT:1> , <EOL> rstype = zxJDBC . TYPE_SCROLL_INSENSITIVE , <EOL> rsconcur = zxJDBC . CONCUR_READ_ONLY <EOL> ) <EOL> self . _test_cursorkeywords ( <NUM_LIT:1> , <EOL> rstype = zxJDBC . TYPE_SCROLL_INSENSITIVE , <EOL> rsconcur = zxJDBC . CONCUR_READ_ONLY <EOL> ) <EOL> self . _test_cursorkeywords ( <NUM_LIT:1> , zxJDBC . TYPE_SCROLL_INSENSITIVE , zxJDBC . CONCUR_READ_ONLY ) <EOL> self . assertRaises ( TypeError , self . cursor , <NUM_LIT:1> , zxJDBC . TYPE_SCROLL_INSENSITIVE ) <EOL> def testFileLikeCursor ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> print >> c , "<STR_LIT>" <EOL> print >> c , "<STR_LIT>" <EOL> print >> c , "<STR_LIT>" <EOL> self . db . commit ( ) <EOL> finally : <EOL> c . close ( ) <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> f = c . fetchall ( ) <EOL> self . assertEquals ( <NUM_LIT:3> , len ( f ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testIteration ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> cnt = <NUM_LIT:0> <EOL> c . execute ( "<STR_LIT>" ) <EOL> for a in c : <EOL> self . assertEquals ( <NUM_LIT:3> , len ( a ) ) <EOL> cnt += <NUM_LIT:1> <EOL> self . assertEquals ( <NUM_LIT:7> , cnt ) <EOL> cnt = <NUM_LIT:0> <EOL> c . execute ( "<STR_LIT>" ) <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> self . assertEquals ( <NUM_LIT:3> , len ( c . next ( ) ) ) <EOL> except StopIteration : <EOL> break <EOL> cnt += <NUM_LIT:1> <EOL> self . assertEquals ( <NUM_LIT:7> , cnt ) <EOL> finally : <EOL> c . close ( ) <EOL> def testClosingCursor ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> finally : <EOL> c . close ( ) <EOL> self . assertRaises ( zxJDBC . ProgrammingError , c . execute , ( "<STR_LIT>" , ) ) <EOL> def testClosingConnectionWithOpenCursors ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> d = self . cursor ( ) <EOL> e = self . cursor ( ) <EOL> self . db . close ( ) <EOL> self . db = self . connect ( ) <EOL> self . assertRaises ( zxJDBC . ProgrammingError , c . execute , ( "<STR_LIT>" , ) ) <EOL> self . assertRaises ( zxJDBC . ProgrammingError , d . execute , ( "<STR_LIT>" , ) ) <EOL> self . assertRaises ( zxJDBC . ProgrammingError , e . execute , ( "<STR_LIT>" , ) ) <EOL> def testNativeSQL ( self ) : <EOL> """<STR_LIT>""" <EOL> sql = self . db . nativesql ( "<STR_LIT>" ) <EOL> assert sql is not None <EOL> assert len ( sql ) > <NUM_LIT:0> <EOL> def testTables ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . tables ( None , None , None , None ) <EOL> found = <NUM_LIT:0> <EOL> while not found : <EOL> try : <EOL> found = "<STR_LIT>" == c . next ( ) [ <NUM_LIT:2> ] . lower ( ) <EOL> except StopIteration : <EOL> break <EOL> assert found , "<STR_LIT>" <EOL> c . tables ( None , None , "<STR_LIT>" , None ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( c . fetchall ( ) ) ) <EOL> c . tables ( None , None , "<STR_LIT>" , ( "<STR_LIT>" , ) ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( c . fetchall ( ) ) ) <EOL> c . tables ( None , None , "<STR_LIT>" , ( "<STR_LIT>" , ) ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( c . fetchall ( ) ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testColumns ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . columns ( None , None , "<STR_LIT>" , None ) <EOL> f = c . fetchall ( ) <EOL> self . assertEquals ( <NUM_LIT:3> , c . rowcount ) <EOL> f . sort ( lambda x , y : cmp ( x [ <NUM_LIT:3> ] , y [ <NUM_LIT:3> ] ) ) <EOL> self . assertEquals ( "<STR_LIT:name>" , f [ <NUM_LIT:1> ] [ <NUM_LIT:3> ] . lower ( ) ) <EOL> if not self . db . __connection__ . getMetaData ( ) . storesMixedCaseIdentifiers ( ) : <EOL> c . columns ( None , None , "<STR_LIT>" , None ) <EOL> f = c . fetchall ( ) <EOL> self . assertEquals ( <NUM_LIT:3> , c . rowcount ) <EOL> f . sort ( lambda x , y : cmp ( x [ <NUM_LIT:3> ] , y [ <NUM_LIT:3> ] ) ) <EOL> self . assertEquals ( "<STR_LIT:name>" , f [ <NUM_LIT:1> ] [ <NUM_LIT:3> ] . lower ( ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testBestRow ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . bestrow ( None , None , "<STR_LIT>" ) <EOL> f = c . fetchall ( ) <EOL> if f : <EOL> self . assertEquals ( <NUM_LIT:1> , len ( f ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testTypeInfo ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . gettypeinfo ( ) <EOL> f = c . fetchall ( ) <EOL> assert f is not None , "<STR_LIT>" <EOL> finally : <EOL> c . close ( ) <EOL> def _test_scrolling ( self , dynamic = <NUM_LIT:0> ) : <EOL> if self . vendor . scroll : <EOL> c = self . cursor ( dynamic , <EOL> rstype = getattr ( zxJDBC , self . vendor . scroll ) , <EOL> rsconcur = zxJDBC . CONCUR_READ_ONLY <EOL> ) <EOL> else : <EOL> c = self . cursor ( dynamic ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . assertEquals ( <NUM_LIT:1> , c . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( <NUM_LIT:2> , c . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( <NUM_LIT:3> , c . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> c . scroll ( - <NUM_LIT:2> ) <EOL> self . assertEquals ( <NUM_LIT:2> , c . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> c . scroll ( <NUM_LIT:4> , "<STR_LIT>" ) <EOL> self . assertEquals ( <NUM_LIT:5> , c . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> c . scroll ( - <NUM_LIT:5> ) <EOL> self . assertEquals ( <NUM_LIT:1> , c . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> c . scroll ( <NUM_LIT:6> , "<STR_LIT>" ) <EOL> self . assertEquals ( <NUM_LIT:7> , c . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertRaises ( IndexError , c . scroll , <NUM_LIT:1> , "<STR_LIT>" ) <EOL> self . assertRaises ( IndexError , c . scroll , - <NUM_LIT:1> , "<STR_LIT>" ) <EOL> self . assertRaises ( zxJDBC . ProgrammingError , c . scroll , <NUM_LIT:1> , "<STR_LIT>" ) <EOL> finally : <EOL> c . close ( ) <EOL> def testDynamicCursorScrolling ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_scrolling ( <NUM_LIT:1> ) <EOL> def testStaticCursorScrolling ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_scrolling ( <NUM_LIT:0> ) <EOL> def _test_rownumber ( self , dynamic = <NUM_LIT:0> ) : <EOL> if self . vendor . scroll : <EOL> c = self . cursor ( dynamic , <EOL> rstype = getattr ( zxJDBC , self . vendor . scroll ) , <EOL> rsconcur = zxJDBC . CONCUR_READ_ONLY <EOL> ) <EOL> else : <EOL> c = self . cursor ( dynamic ) <EOL> try : <EOL> if not dynamic : <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . assertEquals ( <NUM_LIT:0> , c . rownumber ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . assertEquals ( <NUM_LIT:0> , c . rownumber ) <EOL> c . next ( ) <EOL> self . assertEquals ( <NUM_LIT:1> , c . rownumber ) <EOL> c . next ( ) <EOL> self . assertEquals ( <NUM_LIT:2> , c . rownumber ) <EOL> c . scroll ( - <NUM_LIT:1> ) <EOL> self . assertEquals ( <NUM_LIT:1> , c . rownumber ) <EOL> c . scroll ( <NUM_LIT:2> , "<STR_LIT>" ) <EOL> self . assertEquals ( <NUM_LIT:2> , c . rownumber ) <EOL> c . scroll ( <NUM_LIT:6> , "<STR_LIT>" ) <EOL> self . assertEquals ( <NUM_LIT:6> , c . rownumber ) <EOL> finally : <EOL> c . close ( ) <EOL> self . assertEquals ( None , c . rownumber ) <EOL> def testStaticRownumber ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_rownumber ( <NUM_LIT:0> ) <EOL> def testDynamicRownumber ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_rownumber ( <NUM_LIT:1> ) <EOL> def _test_rowcount ( self , dynamic = <NUM_LIT:0> ) : <EOL> if self . vendor . scroll : <EOL> c = self . cursor ( dynamic , <EOL> rstype = getattr ( zxJDBC , self . vendor . scroll ) , <EOL> rsconcur = zxJDBC . CONCUR_READ_ONLY <EOL> ) <EOL> else : <EOL> c = self . cursor ( dynamic ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> c . next ( ) <EOL> c . next ( ) <EOL> c . next ( ) <EOL> if dynamic : <EOL> self . assertEquals ( <NUM_LIT:3> , c . rowcount ) <EOL> else : <EOL> self . assertEquals ( <NUM_LIT:7> , c . rowcount ) <EOL> c . scroll ( - <NUM_LIT:1> ) <EOL> if dynamic : <EOL> self . assertEquals ( <NUM_LIT:3> , c . rowcount ) <EOL> else : <EOL> self . assertEquals ( <NUM_LIT:7> , c . rowcount ) <EOL> finally : <EOL> c . close ( ) <EOL> def testStaticRowcount ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_rowcount ( <NUM_LIT:0> ) <EOL> def testDynamicRowcount ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_rowcount ( <NUM_LIT:1> ) <EOL> def testTableTypeInfo ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . gettabletypeinfo ( ) <EOL> c . fetchall ( ) <EOL> assert c . rowcount > <NUM_LIT:0> , "<STR_LIT>" <EOL> finally : <EOL> c . close ( ) <EOL> def testTupleParams ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> self . assertRaises ( zxJDBC . ProgrammingError , c . execute , "<STR_LIT>" , params = <NUM_LIT:4> ) <EOL> c . execute ( "<STR_LIT>" , params = [ <NUM_LIT:4> ] ) <EOL> c . execute ( "<STR_LIT>" , params = ( <NUM_LIT:4> , ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testConnectionAttribute ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> from com . ziclix . python . sql import PyConnection <EOL> assert isinstance ( c . connection , PyConnection ) , "<STR_LIT>" <EOL> self . assertRaises ( TypeError , setattr , ( c , "<STR_LIT>" , None ) , None ) <EOL> finally : <EOL> c . close ( ) <EOL> def testFetchMany ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> data = c . fetchmany ( <NUM_LIT:6> ) <EOL> assert len ( data ) == <NUM_LIT:6> , "<STR_LIT>" % ( len ( data ) ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> data = c . fetchmany ( <NUM_LIT:16> ) <EOL> assert len ( data ) == <NUM_LIT:7> , "<STR_LIT>" % ( len ( data ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testQueryWithParameter ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" , [ ( "<STR_LIT>" , ) ] , { <NUM_LIT:0> : zxJDBC . VARCHAR } ) <EOL> data = c . fetchall ( ) <EOL> assert len ( data ) == <NUM_LIT:2> , "<STR_LIT>" % ( len ( data ) ) <EOL> c . execute ( "<STR_LIT>" , [ ( "<STR_LIT>" , ) ] , { <NUM_LIT:0> : zxJDBC . VARCHAR } ) <EOL> data = c . fetchall ( ) <EOL> assert len ( data ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( data ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testInsertWithFile ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . has_table ( "<STR_LIT>" ) , "<STR_LIT>" <EOL> fp = open ( tempfile . mktemp ( ) , "<STR_LIT:w>" ) <EOL> c = self . cursor ( ) <EOL> try : <EOL> try : <EOL> c . execute ( self . table ( "<STR_LIT>" ) [ <NUM_LIT:1> ] ) <EOL> data = fp . name * <NUM_LIT> <EOL> data = data [ : <NUM_LIT> ] <EOL> fp . write ( data ) <EOL> fp . flush ( ) <EOL> fp . close ( ) <EOL> fp = open ( fp . name , "<STR_LIT:r>" ) <EOL> c . execute ( "<STR_LIT>" % ( self . table ( "<STR_LIT>" ) [ <NUM_LIT:0> ] ) , [ ( <NUM_LIT:0> , fp ) ] , { <NUM_LIT:1> : zxJDBC . LONGVARCHAR } ) <EOL> self . db . commit ( ) <EOL> c . execute ( "<STR_LIT>" % ( self . table ( "<STR_LIT>" ) [ <NUM_LIT:0> ] ) ) <EOL> f = c . fetchall ( ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( f ) ) <EOL> assert len ( f [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) == len ( data ) , "<STR_LIT>" % ( len ( data ) , len ( f [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) ) <EOL> assert data == f [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] , "<STR_LIT>" <EOL> except Exception , e : <EOL> raise e <EOL> finally : <EOL> c . execute ( "<STR_LIT>" % ( self . table ( "<STR_LIT>" ) [ <NUM_LIT:0> ] ) ) <EOL> c . close ( ) <EOL> self . db . commit ( ) <EOL> fp . close ( ) <EOL> os . remove ( fp . name ) <EOL> def calendar ( self ) : <EOL> c = Calendar . getInstance ( ) <EOL> c . setTime ( JDate ( ) ) <EOL> return c <EOL> def testDate ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . calendar ( ) <EOL> o = zxJDBC . DateFromTicks ( c . getTime ( ) . getTime ( ) / <NUM_LIT:1000> L ) <EOL> v = zxJDBC . Date ( c . get ( Calendar . YEAR ) , c . get ( Calendar . MONTH ) + <NUM_LIT:1> , c . get ( Calendar . DATE ) ) <EOL> assert o . equals ( v ) , "<STR_LIT>" % ( v . getTime ( ) , o . getTime ( ) ) <EOL> t = time . time ( ) <EOL> l = time . localtime ( t ) <EOL> o = zxJDBC . DateFromTicks ( t ) <EOL> v = zxJDBC . Date ( l [ <NUM_LIT:0> ] , l [ <NUM_LIT:1> ] , l [ <NUM_LIT:2> ] ) <EOL> assert o . equals ( v ) , "<STR_LIT>" % ( v . getTime ( ) , o . getTime ( ) ) <EOL> def testTime ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . calendar ( ) <EOL> o = zxJDBC . TimeFromTicks ( c . getTime ( ) . getTime ( ) / <NUM_LIT:1000> L ) <EOL> v = zxJDBC . Time ( c . get ( Calendar . HOUR ) , c . get ( Calendar . MINUTE ) , c . get ( Calendar . SECOND ) ) <EOL> assert o . equals ( v ) , "<STR_LIT>" % ( v . getTime ( ) , o . getTime ( ) ) <EOL> def testTimestamp ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . calendar ( ) <EOL> o = zxJDBC . TimestampFromTicks ( c . getTime ( ) . getTime ( ) / <NUM_LIT:1000> L ) <EOL> v = zxJDBC . Timestamp ( c . get ( Calendar . YEAR ) , c . get ( Calendar . MONTH ) + <NUM_LIT:1> , c . get ( Calendar . DATE ) , <EOL> c . get ( Calendar . HOUR ) , c . get ( Calendar . MINUTE ) , c . get ( Calendar . SECOND ) ) <EOL> assert o . equals ( v ) , "<STR_LIT>" % ( v . getTime ( ) , o . getTime ( ) ) <EOL> def _test_precision ( self , ( tabname , sql ) , diff , values , attr ) : <EOL> try : <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" % ( tabname ) ) <EOL> self . db . commit ( ) <EOL> except : <EOL> self . db . rollback ( ) <EOL> finally : <EOL> c . close ( ) <EOL> try : <EOL> c = self . cursor ( ) <EOL> c . execute ( sql ) <EOL> c . execute ( "<STR_LIT>" % ( tabname ) , map ( lambda x : ( <NUM_LIT:0> , x ) , values ) ) <EOL> c . execute ( "<STR_LIT>" % ( tabname ) ) <EOL> f = c . fetchall ( ) <EOL> assert len ( values ) == len ( f ) , "<STR_LIT>" <EOL> for i in range ( <NUM_LIT:0> , len ( f ) ) : <EOL> v = values [ i ] <EOL> if attr : v = getattr ( v , attr ) ( ) <EOL> msg = "<STR_LIT>" % ( v , f [ i ] [ <NUM_LIT:1> ] , ( i + <NUM_LIT:1> ) , len ( f ) ) <EOL> assert diff ( f [ i ] [ <NUM_LIT:1> ] , values [ i ] ) < <NUM_LIT> , msg <EOL> self . db . commit ( ) <EOL> finally : <EOL> c . close ( ) <EOL> try : <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" % ( tabname ) ) <EOL> self . db . commit ( ) <EOL> except : <EOL> self . db . rollback ( ) <EOL> finally : <EOL> c . close ( ) <EOL> def testFloat ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . has_table ( "<STR_LIT>" ) , "<STR_LIT>" <EOL> values = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] <EOL> self . _test_precision ( self . table ( "<STR_LIT>" ) , lambda x , y : x - y , values , None ) <EOL> def testBigDecimal ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . has_table ( "<STR_LIT>" ) , "<STR_LIT>" <EOL> from java . math import BigDecimal <EOL> values = [ BigDecimal ( x ) . setScale ( <NUM_LIT:2> , BigDecimal . ROUND_UP ) for x in [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ] <EOL> self . _test_precision ( self . table ( "<STR_LIT>" ) , lambda x , y , b = BigDecimal : b ( x ) . subtract ( y ) . doubleValue ( ) , values , "<STR_LIT>" ) <EOL> def testBigDecimalConvertedToDouble ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . has_table ( "<STR_LIT>" ) , "<STR_LIT>" <EOL> from java . math import BigDecimal <EOL> values = [ BigDecimal ( x ) . setScale ( <NUM_LIT:2> , BigDecimal . ROUND_UP ) for x in [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ] <EOL> self . _test_precision ( self . table ( "<STR_LIT>" ) , lambda x , y : x - y . doubleValue ( ) , values , "<STR_LIT>" ) <EOL> def testNextset ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" , [ ( <NUM_LIT:3> , ) , ( <NUM_LIT:4> , ) ] ) <EOL> f = c . fetchall ( ) <EOL> assert f , "<STR_LIT>" <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( f ) ) <EOL> assert c . nextset ( ) , "<STR_LIT>" <EOL> f = c . fetchall ( ) <EOL> assert f , "<STR_LIT>" <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( f ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testJavaUtilList ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> from java . util import LinkedList <EOL> a = LinkedList ( ) <EOL> a . add ( ( <NUM_LIT:3> , ) ) <EOL> c . execute ( "<STR_LIT>" , a ) <EOL> f = c . fetchall ( ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( f ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testUpdateCount ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" , [ ( <NUM_LIT> , '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> assert c . updatecount == <NUM_LIT:1> , "<STR_LIT>" % ( c . updatecount ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . assertEquals ( None , c . updatecount ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> assert c . updatecount == <NUM_LIT:8> , "<STR_LIT>" % ( c . updatecount ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . assertEquals ( <NUM_LIT:0> , c . updatecount ) <EOL> finally : <EOL> c . close ( ) <EOL> def _test_time ( self , ( tabname , sql ) , factory , values , _type , _cmp = cmp , datahandler = None ) : <EOL> c = self . cursor ( ) <EOL> if datahandler : c . datahandler = datahandler ( c . datahandler ) <EOL> try : <EOL> c . execute ( sql ) <EOL> dates = map ( lambda x , f = factory : apply ( f , x ) , values ) <EOL> for a in dates : <EOL> c . execute ( "<STR_LIT>" % ( tabname ) , [ ( a , ) ] , { <NUM_LIT:0> : _type } ) <EOL> self . db . commit ( ) <EOL> c . execute ( "<STR_LIT>" % ( tabname ) , [ ( dates [ <NUM_LIT:0> ] , ) ] , { <NUM_LIT:0> : _type } ) <EOL> f = c . fetchall ( ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( f ) ) <EOL> assert _cmp ( f [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] , dates [ <NUM_LIT:0> ] ) == <NUM_LIT:0> , "<STR_LIT>" % ( str ( dates [ <NUM_LIT:0> ] ) , str ( f [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) ) <EOL> c . execute ( "<STR_LIT>" % ( tabname ) , [ ( dates [ <NUM_LIT:1> ] , ) ] , { <NUM_LIT:0> : _type } ) <EOL> self . db . commit ( ) <EOL> c . execute ( "<STR_LIT>" % ( tabname ) ) <EOL> f = c . fetchall ( ) <EOL> self . assertEquals ( len ( f ) , len ( dates ) - <NUM_LIT:1> ) <EOL> finally : <EOL> c . execute ( "<STR_LIT>" % ( tabname ) ) <EOL> c . close ( ) <EOL> self . db . commit ( ) <EOL> def testUpdateSelectByDate ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . has_table ( "<STR_LIT>" ) , "<STR_LIT>" <EOL> def _cmp_ ( x , y ) : <EOL> xt = ( x . getYear ( ) , x . getMonth ( ) , x . getDay ( ) ) <EOL> yt = ( y . getYear ( ) , y . getMonth ( ) , y . getDay ( ) ) <EOL> return not xt == yt <EOL> values = [ ( <NUM_LIT> , <NUM_LIT:6> , <NUM_LIT> ) , ( <NUM_LIT> , <NUM_LIT:11> , <NUM_LIT:12> ) , ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:12> ) , ( <NUM_LIT> , <NUM_LIT:9> , <NUM_LIT> ) ] <EOL> self . _test_time ( self . table ( "<STR_LIT>" ) , zxJDBC . Date , values , zxJDBC . DATE , _cmp_ ) <EOL> def testUpdateSelectByTime ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . has_table ( "<STR_LIT>" ) , "<STR_LIT>" <EOL> def _cmp_ ( x , y ) : <EOL> xt = ( x . getHours ( ) , x . getMinutes ( ) , x . getSeconds ( ) ) <EOL> yt = ( y . getHours ( ) , y . getMinutes ( ) , y . getSeconds ( ) ) <EOL> return not xt == yt <EOL> values = [ ( <NUM_LIT:10> , <NUM_LIT:11> , <NUM_LIT:12> ) , ( <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:12> ) , ( <NUM_LIT> , <NUM_LIT:9> , <NUM_LIT> ) ] <EOL> self . _test_time ( self . table ( "<STR_LIT>" ) , zxJDBC . Time , values , zxJDBC . TIME , _cmp_ ) <EOL> def testUpdateSelectByTimestamp ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . has_table ( "<STR_LIT>" ) , "<STR_LIT>" <EOL> def _cmp_ ( x , y ) : <EOL> xt = ( x . getYear ( ) , x . getMonth ( ) , x . getDay ( ) , x . getHours ( ) , x . getMinutes ( ) , x . getSeconds ( ) ) <EOL> yt = ( y . getYear ( ) , y . getMonth ( ) , y . getDay ( ) , y . getHours ( ) , y . getMinutes ( ) , y . getSeconds ( ) ) <EOL> return not xt == yt <EOL> values = [ ( <NUM_LIT> , <NUM_LIT:6> , <NUM_LIT> , <NUM_LIT:10> , <NUM_LIT:11> , <NUM_LIT:12> ) , ( <NUM_LIT> , <NUM_LIT:11> , <NUM_LIT:12> , <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:12> ) , ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:12> , <NUM_LIT:4> , <NUM_LIT:9> , <NUM_LIT> ) ] <EOL> self . _test_time ( self . table ( "<STR_LIT>" ) , zxJDBC . Timestamp , values , zxJDBC . TIMESTAMP , _cmp_ ) <EOL> def testOrderOfArgsMaxRowsOnly ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" , maxrows = <NUM_LIT:3> ) <EOL> f = c . fetchall ( ) <EOL> assert len ( f ) == <NUM_LIT:3> , "<STR_LIT>" % ( len ( f ) ) <EOL> finally : <EOL> c . close ( ) <EOL> self . db . commit ( ) <EOL> def testOrderOfArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" , bindings = { <NUM_LIT:0> : zxJDBC . INTEGER } , params = [ ( <NUM_LIT:3> , ) ] ) <EOL> f = c . fetchall ( ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( f ) ) <EOL> c . execute ( "<STR_LIT>" , bindings = { } , params = [ ( <NUM_LIT:3> , ) ] ) <EOL> f = c . fetchall ( ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( f ) ) <EOL> c . execute ( "<STR_LIT>" , bindings = { } , params = [ ] ) <EOL> f = c . fetchall ( ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( f ) ) <EOL> finally : <EOL> c . close ( ) <EOL> self . db . commit ( ) <EOL> def testMaxrows ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" , maxrows = <NUM_LIT:3> ) <EOL> self . assertEquals ( <NUM_LIT:3> , len ( c . fetchall ( ) ) ) <EOL> c . execute ( "<STR_LIT>" , ( <NUM_LIT:1> , ) , maxrows = <NUM_LIT:3> ) <EOL> self . assertEquals ( <NUM_LIT:3> , len ( c . fetchall ( ) ) ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> f = c . fetchall ( ) <EOL> num = f [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> c . execute ( "<STR_LIT>" , maxrows = <NUM_LIT:0> ) <EOL> self . assertEquals ( num , len ( c . fetchall ( ) ) ) <EOL> finally : <EOL> c . close ( ) <EOL> self . db . commit ( ) <EOL> def testPrimaryKey ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . primarykeys ( None , None , "<STR_LIT>" ) <EOL> f = c . fetchall ( ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( f ) ) <EOL> assert f [ <NUM_LIT:0> ] [ <NUM_LIT:3> ] . lower ( ) == "<STR_LIT:id>" , "<STR_LIT>" % ( f [ <NUM_LIT:0> ] [ <NUM_LIT:3> ] ) <EOL> finally : <EOL> c . close ( ) <EOL> self . db . commit ( ) <EOL> def testForeignKey ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def testIndexInfo ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . statistics ( None , None , "<STR_LIT>" , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> f = c . fetchall ( ) <EOL> assert f is not None , "<STR_LIT>" <EOL> f = filter ( lambda x : x [ <NUM_LIT:5> ] , f ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( f ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def _test_fetching ( self , dynamic = <NUM_LIT:0> ) : <EOL> c = self . cursor ( dynamic ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . assertEquals ( None , c . fetchone ( ) ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . assertEquals ( [ ] , c . fetchmany ( ) ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . assertEquals ( [ ] , c . fetchall ( ) ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> f = c . fetchmany ( ) <EOL> assert len ( f ) == c . arraysize , "<STR_LIT>" % ( c . arraysize , len ( f ) ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> c . arraysize = <NUM_LIT:4> <EOL> f = c . fetchmany ( ) <EOL> assert len ( f ) == <NUM_LIT:4> , "<STR_LIT>" % ( len ( f ) ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> c . arraysize = - <NUM_LIT:1> <EOL> f = c . fetchmany ( ) <EOL> assert len ( f ) == <NUM_LIT:7> , "<STR_LIT>" % ( len ( f ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testStaticFetching ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_fetching ( <NUM_LIT:0> ) <EOL> def testDynamicFetching ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_fetching ( <NUM_LIT:1> ) <EOL> def testFetchingBeforeExecute ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> try : <EOL> c . fetchall ( ) <EOL> except zxJDBC . Error , e : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> finally : <EOL> c . close ( ) <EOL> def testBindingsWithNoParams ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( ) <EOL> try : <EOL> self . assertRaises ( zxJDBC . ProgrammingError , c . execute , "<STR_LIT>" , { <NUM_LIT:0> : zxJDBC . INTEGER } ) <EOL> self . assertRaises ( zxJDBC . ProgrammingError , c . execute , "<STR_LIT>" , { <NUM_LIT:0> : { } } ) <EOL> finally : <EOL> c . close ( ) <EOL> def testDynamicCursor ( self ) : <EOL> """<STR_LIT>""" <EOL> c = self . cursor ( <NUM_LIT:1> ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> f = c . fetchmany ( <NUM_LIT:4> ) <EOL> assert len ( f ) == <NUM_LIT:4> , "<STR_LIT>" % ( len ( f ) ) <EOL> finally : <EOL> c . close ( ) <EOL> def testRowid ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . has_table ( "<STR_LIT>" ) , "<STR_LIT>" <EOL> c = self . cursor ( ) <EOL> assert c . lastrowid == None , "<STR_LIT>" <EOL> try : <EOL> tabname , sql = self . table ( "<STR_LIT>" ) <EOL> c . execute ( sql ) <EOL> c . execute ( "<STR_LIT>" % ( tabname ) , [ ( <NUM_LIT:0> , ) ] ) <EOL> assert c . lastrowid is not None , "<STR_LIT>" <EOL> try : <EOL> for idx in range ( c . lastrowid + <NUM_LIT:1> , c . lastrowid + <NUM_LIT> ) : <EOL> c . execute ( "<STR_LIT>" % ( tabname ) , [ ( idx , ) ] ) <EOL> assert c . lastrowid is not None , "<STR_LIT>" <EOL> self . assertEquals ( idx , c . lastrowid ) <EOL> except : <EOL> self . db . rollback ( ) <EOL> finally : <EOL> if self . has_table ( "<STR_LIT>" ) : <EOL> try : <EOL> sequence , sql = self . table ( "<STR_LIT>" ) <EOL> c . execute ( sql ) <EOL> self . db . commit ( ) <EOL> except : <EOL> self . db . rollback ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" % ( tabname ) ) <EOL> self . db . commit ( ) <EOL> except : <EOL> self . db . rollback ( ) <EOL> self . db . commit ( ) <EOL> c . close ( ) <EOL> def _test_fetchapi ( self , dynamic = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> from com . ziclix . python . sql import Fetch , WarningListener <EOL> cur = self . cursor ( ) <EOL> try : <EOL> c = self . db . __connection__ <EOL> stmt = c . prepareStatement ( "<STR_LIT>" ) <EOL> stmt . setInt ( <NUM_LIT:1> , <NUM_LIT:5> ) <EOL> rs = stmt . executeQuery ( ) <EOL> fetch = Fetch . newFetch ( cur . datahandler , dynamic ) <EOL> class WL ( WarningListener ) : <EOL> def warning ( self , event ) : <EOL> raise event . getWarning ( ) <EOL> wl = WL ( ) <EOL> fetch . addWarningListener ( wl ) <EOL> fetch . add ( rs ) <EOL> if not dynamic : <EOL> self . assertEquals ( <NUM_LIT:4> , fetch . getRowCount ( ) ) <EOL> assert fetch . fetchone ( ) <EOL> assert fetch . fetchmany ( <NUM_LIT:2> ) <EOL> assert fetch . fetchall ( ) <EOL> assert not fetch . fetchall ( ) <EOL> self . assertEquals ( <NUM_LIT:4> , fetch . getRowCount ( ) ) <EOL> assert fetch . removeWarningListener ( wl ) <EOL> fetch . close ( ) <EOL> stmt . close ( ) <EOL> finally : <EOL> cur . close ( ) <EOL> def testStaticFetchAPI ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_fetchapi ( <NUM_LIT:0> ) <EOL> def testDynamicFetchAPI ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_fetchapi ( <NUM_LIT:1> ) <EOL> class LOBTestCase ( zxJDBCTestCase ) : <EOL> def _test_blob ( self , obj = <NUM_LIT:0> ) : <EOL> assert self . has_table ( "<STR_LIT>" ) , "<STR_LIT>" <EOL> tabname , sql = self . table ( "<STR_LIT>" ) <EOL> fn = tempfile . mktemp ( ) <EOL> fp = None <EOL> c = self . cursor ( ) <EOL> try : <EOL> hello = ( "<STR_LIT:hello>" , ) * <NUM_LIT> <EOL> c . execute ( sql ) <EOL> self . db . commit ( ) <EOL> from java . io import FileOutputStream , FileInputStream , ObjectOutputStream , ObjectInputStream , ByteArrayInputStream <EOL> fp = FileOutputStream ( fn ) <EOL> oos = ObjectOutputStream ( fp ) <EOL> oos . writeObject ( hello ) <EOL> fp . close ( ) <EOL> fp = FileInputStream ( fn ) <EOL> blob = ObjectInputStream ( fp ) <EOL> value = blob . readObject ( ) <EOL> fp . close ( ) <EOL> assert hello == value , "<STR_LIT>" <EOL> if obj == <NUM_LIT:1> : <EOL> fp = open ( fn , "<STR_LIT:rb>" ) <EOL> else : <EOL> fp = FileInputStream ( fn ) <EOL> c . execute ( "<STR_LIT>" % ( tabname ) , [ ( <NUM_LIT:0> , fp ) ] , { <NUM_LIT:1> : zxJDBC . BLOB } ) <EOL> self . db . commit ( ) <EOL> c . execute ( "<STR_LIT>" % ( tabname ) ) <EOL> f = c . fetchall ( ) <EOL> bytes = f [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> blob = ObjectInputStream ( ByteArrayInputStream ( bytes ) ) . readObject ( ) <EOL> assert hello == blob , "<STR_LIT>" <EOL> finally : <EOL> c . execute ( "<STR_LIT>" % ( tabname ) ) <EOL> c . close ( ) <EOL> self . db . commit ( ) <EOL> if os . path . exists ( fn ) : <EOL> if fp : <EOL> fp . close ( ) <EOL> os . remove ( fn ) <EOL> def testBLOBAsString ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_blob ( ) <EOL> def testBLOBAsPyFile ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_blob ( <NUM_LIT:1> ) <EOL> def _test_clob ( self , asfile = <NUM_LIT:0> ) : <EOL> assert self . has_table ( "<STR_LIT>" ) , "<STR_LIT>" <EOL> tabname , sql = self . table ( "<STR_LIT>" ) <EOL> c = self . cursor ( ) <EOL> try : <EOL> hello = "<STR_LIT:hello>" * <NUM_LIT> * <NUM_LIT:10> <EOL> c . execute ( sql ) <EOL> self . db . commit ( ) <EOL> if asfile : <EOL> fp = open ( tempfile . mktemp ( ) , "<STR_LIT:w>" ) <EOL> fp . write ( hello ) <EOL> fp . flush ( ) <EOL> fp . close ( ) <EOL> obj = open ( fp . name , "<STR_LIT:r>" ) <EOL> else : <EOL> obj = hello <EOL> c . execute ( "<STR_LIT>" % ( tabname ) , [ ( <NUM_LIT:0> , obj ) ] , { <NUM_LIT:1> : zxJDBC . CLOB } ) <EOL> c . execute ( "<STR_LIT>" % ( tabname ) , maxrows = <NUM_LIT:1> ) <EOL> f = c . fetchall ( ) <EOL> assert len ( f ) == <NUM_LIT:1> , "<STR_LIT>" % ( <NUM_LIT:1> , len ( f ) ) <EOL> assert hello == f [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] , "<STR_LIT>" <EOL> finally : <EOL> c . execute ( "<STR_LIT>" % ( tabname ) ) <EOL> c . close ( ) <EOL> self . db . commit ( ) <EOL> if asfile : <EOL> obj . close ( ) <EOL> os . remove ( obj . name ) <EOL> def testCLOBAsString ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_clob ( <NUM_LIT:0> ) <EOL> def testCLOBAsPyFile ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_clob ( <NUM_LIT:1> ) <EOL> class BCPTestCase ( zxJDBCTestCase ) : <EOL> def testCSVPipe ( self ) : <EOL> """<STR_LIT>""" <EOL> from java . io import PrintWriter , FileWriter <EOL> from com . ziclix . python . sql . pipe import Pipe <EOL> from com . ziclix . python . sql . pipe . db import DBSource <EOL> from com . ziclix . python . sql . pipe . csv import CSVSink <EOL> try : <EOL> src = self . connect ( ) <EOL> fn = tempfile . mktemp ( suffix = "<STR_LIT>" ) <EOL> writer = PrintWriter ( FileWriter ( fn ) ) <EOL> csvSink = CSVSink ( writer ) <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" , [ ( <NUM_LIT:1000> , '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> c . execute ( "<STR_LIT>" , [ ( <NUM_LIT> , '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> c . execute ( "<STR_LIT>" , [ ( <NUM_LIT> , '<STR_LIT>' , None ) ] , { <NUM_LIT:2> : zxJDBC . VARCHAR } ) <EOL> self . db . commit ( ) <EOL> finally : <EOL> self . db . rollback ( ) <EOL> c . close ( ) <EOL> dbSource = DBSource ( src , c . datahandler . __class__ , "<STR_LIT>" , None , None , None ) <EOL> cnt = Pipe ( ) . pipe ( dbSource , csvSink ) - <NUM_LIT:1> <EOL> finally : <EOL> writer . close ( ) <EOL> src . close ( ) <EOL> os . remove ( fn ) <EOL> def testDBPipe ( self ) : <EOL> """<STR_LIT>""" <EOL> from com . ziclix . python . sql . pipe import Pipe <EOL> from com . ziclix . python . sql . pipe . db import DBSource , DBSink <EOL> try : <EOL> src = self . connect ( ) <EOL> dst = self . connect ( ) <EOL> c = self . cursor ( ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . db . commit ( ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> one = c . fetchone ( ) [ <NUM_LIT:0> ] <EOL> c . close ( ) <EOL> dbSource = DBSource ( src , c . datahandler . __class__ , "<STR_LIT>" , None , None , None ) <EOL> dbSink = DBSink ( dst , c . datahandler . __class__ , "<STR_LIT>" , None , None , <NUM_LIT:1> ) <EOL> cnt = Pipe ( ) . pipe ( dbSource , dbSink ) - <NUM_LIT:1> <EOL> c = self . cursor ( ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> two = c . fetchone ( ) [ <NUM_LIT:0> ] <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . db . commit ( ) <EOL> c . close ( ) <EOL> assert one == two , "<STR_LIT>" % ( one , two ) <EOL> assert one == cnt , "<STR_LIT>" % ( one , cnt ) <EOL> dbSource = DBSource ( src , c . datahandler . __class__ , "<STR_LIT>" , None , [ "<STR_LIT:id>" ] , None ) <EOL> dbSink = DBSink ( dst , c . datahandler . __class__ , "<STR_LIT>" , [ "<STR_LIT:id>" ] , None , <NUM_LIT:1> ) <EOL> self . assertRaises ( zxJDBC . Error , Pipe ( ) . pipe , dbSource , dbSink ) <EOL> params = [ ( <NUM_LIT:4> , ) ] <EOL> dbSource = DBSource ( src , c . datahandler . __class__ , "<STR_LIT>" , "<STR_LIT>" , None , params ) <EOL> dbSink = DBSink ( dst , c . datahandler . __class__ , "<STR_LIT>" , None , None , <NUM_LIT:1> ) <EOL> cnt = Pipe ( ) . pipe ( dbSource , dbSink ) - <NUM_LIT:1> <EOL> c = self . cursor ( ) <EOL> c . execute ( "<STR_LIT>" , params ) <EOL> one = c . fetchone ( ) [ <NUM_LIT:0> ] <EOL> c . execute ( "<STR_LIT>" ) <EOL> two = c . fetchone ( ) [ <NUM_LIT:0> ] <EOL> c . close ( ) <EOL> assert one == two , "<STR_LIT>" % ( one , two ) <EOL> assert one == cnt , "<STR_LIT>" % ( one , cnt ) <EOL> finally : <EOL> try : <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . db . commit ( ) <EOL> except : <EOL> self . db . rollback ( ) <EOL> finally : <EOL> c . close ( ) <EOL> try : <EOL> src . close ( ) <EOL> except : <EOL> src = None <EOL> try : <EOL> dst . close ( ) <EOL> except : <EOL> dst = None <EOL> def testBCP ( self ) : <EOL> """<STR_LIT>""" <EOL> from com . ziclix . python . sql . util import BCP <EOL> import dbexts <EOL> try : <EOL> src = self . connect ( ) <EOL> dst = self . connect ( ) <EOL> c = self . cursor ( ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . db . commit ( ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> one = c . fetchone ( ) [ <NUM_LIT:0> ] <EOL> c . close ( ) <EOL> b = BCP ( src , dst ) <EOL> if hasattr ( self , "<STR_LIT>" ) : <EOL> b . sourceDataHandler = self . datahandler <EOL> b . destinationDataHandler = self . datahandler <EOL> cnt = b . bcp ( "<STR_LIT>" , toTable = "<STR_LIT>" ) <EOL> c = self . cursor ( ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> two = c . fetchone ( ) [ <NUM_LIT:0> ] <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . db . commit ( ) <EOL> c . close ( ) <EOL> assert one == two , "<STR_LIT>" % ( one , two ) <EOL> assert one == cnt , "<STR_LIT>" % ( one , cnt ) <EOL> self . assertRaises ( zxJDBC . Error , b . bcp , "<STR_LIT>" , toTable = "<STR_LIT>" , include = [ "<STR_LIT:id>" ] , exclude = [ "<STR_LIT:id>" ] ) <EOL> params = [ ( <NUM_LIT:4> , ) ] <EOL> cnt = b . bcp ( "<STR_LIT>" , "<STR_LIT>" , params , toTable = "<STR_LIT>" ) <EOL> c = self . cursor ( ) <EOL> c . execute ( "<STR_LIT>" , params ) <EOL> one = c . fetchone ( ) [ <NUM_LIT:0> ] <EOL> c . execute ( "<STR_LIT>" ) <EOL> two = c . fetchone ( ) [ <NUM_LIT:0> ] <EOL> c . close ( ) <EOL> assert one == two , "<STR_LIT>" % ( one , two ) <EOL> assert one == cnt , "<STR_LIT>" % ( one , cnt ) <EOL> finally : <EOL> try : <EOL> c = self . cursor ( ) <EOL> try : <EOL> c . execute ( "<STR_LIT>" ) <EOL> self . db . commit ( ) <EOL> except : <EOL> self . db . rollback ( ) <EOL> finally : <EOL> c . close ( ) <EOL> try : <EOL> src . close ( ) <EOL> except : <EOL> src = None <EOL> try : <EOL> dst . close ( ) <EOL> except : <EOL> dst = None </s>
