<s> """<STR_LIT>""" <EOL> import random <EOL> from twisted . internet import reactor , interfaces , defer <EOL> from twisted . enterprise . row import RowObject <EOL> from twisted . enterprise . reflector import EQUAL <EOL> from twisted . enterprise . sqlreflector import SQLReflector <EOL> from twisted . enterprise import util <EOL> from twisted . test . test_adbapi import makeSQLTests <EOL> from twisted . trial . util import suppress as suppressWarning <EOL> from twisted . trial import unittest <EOL> tableName = "<STR_LIT>" <EOL> childTableName = "<STR_LIT>" <EOL> class TestRow ( RowObject ) : <EOL> rowColumns = [ ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:int>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:int>" ) ] <EOL> rowKeyColumns = [ ( "<STR_LIT>" , "<STR_LIT>" ) ] <EOL> rowTableName = tableName <EOL> class ChildRow ( RowObject ) : <EOL> rowColumns = [ ( "<STR_LIT>" , "<STR_LIT:int>" ) , <EOL> ( "<STR_LIT:foo>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:int>" ) , <EOL> ( "<STR_LIT:data>" , "<STR_LIT>" ) ] <EOL> rowKeyColumns = [ ( "<STR_LIT>" , "<STR_LIT:int>" ) ] <EOL> rowTableName = childTableName <EOL> rowForeignKeys = [ ( tableName , <EOL> [ ( "<STR_LIT>" , "<STR_LIT>" ) ] , <EOL> [ ( "<STR_LIT>" , "<STR_LIT>" ) ] , <EOL> None , <NUM_LIT:1> ) ] <EOL> main_table_schema = """<STR_LIT>""" <EOL> child_table_schema = """<STR_LIT>""" <EOL> def randomizeRow ( row , nulls_ok = True , trailing_spaces_ok = True ) : <EOL> values = { } <EOL> for name , type in row . rowColumns : <EOL> if util . getKeyColumn ( row , name ) : <EOL> values [ name ] = getattr ( row , name ) <EOL> continue <EOL> elif nulls_ok and random . randint ( <NUM_LIT:0> , <NUM_LIT:9> ) == <NUM_LIT:0> : <EOL> value = None <EOL> elif type == '<STR_LIT:int>' : <EOL> value = random . randint ( - <NUM_LIT> , <NUM_LIT> ) <EOL> else : <EOL> if random . randint ( <NUM_LIT:0> , <NUM_LIT:9> ) == <NUM_LIT:0> : <EOL> value = '<STR_LIT>' <EOL> else : <EOL> value = '<STR_LIT>' . join ( map ( lambda i : chr ( random . randrange ( <NUM_LIT:32> , <NUM_LIT> ) ) , <EOL> xrange ( random . randint ( <NUM_LIT:1> , <NUM_LIT:64> ) ) ) ) <EOL> if not trailing_spaces_ok : <EOL> value = value . rstrip ( ) <EOL> setattr ( row , name , value ) <EOL> values [ name ] = value <EOL> return values <EOL> def rowMatches ( row , values ) : <EOL> for name , type in row . rowColumns : <EOL> if getattr ( row , name ) != values [ name ] : <EOL> print ( "<STR_LIT>" % <EOL> ( name , getattr ( row , name ) , values [ name ] ) ) <EOL> return False <EOL> return True <EOL> rowObjectSuppression = suppressWarning ( <EOL> message = "<STR_LIT>" , <EOL> category = DeprecationWarning ) <EOL> reflectorSuppression = suppressWarning ( <EOL> message = "<STR_LIT>" , <EOL> category = DeprecationWarning ) <EOL> class ReflectorTestBase : <EOL> """<STR_LIT>""" <EOL> if interfaces . IReactorThreads ( reactor , None ) is None : <EOL> skip = "<STR_LIT>" <EOL> count = <NUM_LIT:100> <EOL> def randomizeRow ( self , row ) : <EOL> return randomizeRow ( row , self . nulls_ok , self . trailing_spaces_ok ) <EOL> def extraSetUp ( self ) : <EOL> """<STR_LIT>""" <EOL> d = self . createReflector ( ) <EOL> d . addCallback ( self . _cbSetUp ) <EOL> return d <EOL> def _cbSetUp ( self , reflector ) : <EOL> self . reflector = reflector <EOL> def tearDown ( self ) : <EOL> return self . destroyReflector ( ) <EOL> def destroyReflector ( self ) : <EOL> pass <EOL> def test_reflector ( self ) : <EOL> """<STR_LIT>""" <EOL> row = TestRow ( ) <EOL> row . assignKeyAttr ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> values = self . randomizeRow ( row ) <EOL> d = self . reflector . insertRow ( row ) <EOL> def _loadBack ( _ ) : <EOL> whereClause = [ ( "<STR_LIT>" , EQUAL , "<STR_LIT>" ) ] <EOL> d = self . reflector . loadObjectsFrom ( tableName , <EOL> whereClause = whereClause ) <EOL> return d . addCallback ( self . gotData ) <EOL> def _getParent ( _ ) : <EOL> self . failUnless ( len ( self . data ) == <NUM_LIT:1> , "<STR_LIT>" ) <EOL> parent = self . data [ <NUM_LIT:0> ] <EOL> self . failUnless ( rowMatches ( parent , values ) , "<STR_LIT>" ) <EOL> return parent <EOL> d . addCallback ( _loadBack ) <EOL> d . addCallback ( _getParent ) <EOL> d . addCallback ( self . _cbTestReflector ) <EOL> return d <EOL> test_reflector . suppress = [ rowObjectSuppression , reflectorSuppression ] <EOL> def _cbTestReflector ( self , parent ) : <EOL> test_values = { } <EOL> inserts = [ ] <EOL> child_values = { } <EOL> for i in range ( <NUM_LIT:0> , self . num_iterations ) : <EOL> row = ChildRow ( ) <EOL> row . assignKeyAttr ( "<STR_LIT>" , i ) <EOL> values = self . randomizeRow ( row ) <EOL> values [ '<STR_LIT>' ] = row . test_key = "<STR_LIT>" <EOL> child_values [ i ] = values <EOL> inserts . append ( self . reflector . insertRow ( row ) ) <EOL> row = None <EOL> d = defer . gatherResults ( inserts ) <EOL> values = [ None ] <EOL> def _loadObjects ( _ ) : <EOL> d = self . reflector . loadObjectsFrom ( childTableName , parentRow = parent ) <EOL> return d . addCallback ( self . gotData ) <EOL> def _checkLoadObjects ( _ ) : <EOL> self . failUnless ( len ( self . data ) == self . num_iterations , <EOL> "<STR_LIT>" ) <EOL> self . failUnless ( len ( parent . childRows ) == self . num_iterations , <EOL> "<STR_LIT>" % len ( parent . childRows ) ) <EOL> for child in parent . childRows : <EOL> self . failUnless ( rowMatches ( child , child_values [ child . childId ] ) , <EOL> "<STR_LIT>" % child . childId ) <EOL> def _checkLoadObjects2 ( _ ) : <EOL> self . failUnless ( len ( self . data ) == self . num_iterations , <EOL> "<STR_LIT>" ) <EOL> self . failUnless ( len ( parent . childRows ) == self . num_iterations , <EOL> "<STR_LIT>" % len ( parent . childRows ) ) <EOL> def _changeParent ( _ ) : <EOL> values [ <NUM_LIT:0> ] = self . randomizeRow ( parent ) <EOL> return self . reflector . updateRow ( parent ) <EOL> def _loadBack ( _ ) : <EOL> whereClause = [ ( "<STR_LIT>" , EQUAL , "<STR_LIT>" ) ] <EOL> d = self . reflector . loadObjectsFrom ( tableName , whereClause = whereClause ) <EOL> return d . addCallback ( self . gotData ) <EOL> def _checkLoadBack ( _ ) : <EOL> self . failUnless ( len ( self . data ) == <NUM_LIT:1> , "<STR_LIT>" ) <EOL> parent = self . data [ <NUM_LIT:0> ] <EOL> self . failUnless ( rowMatches ( parent , values [ <NUM_LIT:0> ] ) , "<STR_LIT>" ) <EOL> test_values [ parent . key_string ] = values [ <NUM_LIT:0> ] <EOL> parent = None <EOL> def _saveMoreTestRows ( _ ) : <EOL> ds = [ ] <EOL> for i in range ( <NUM_LIT:0> , self . num_iterations ) : <EOL> row = TestRow ( ) <EOL> row . assignKeyAttr ( "<STR_LIT>" , "<STR_LIT>" % i ) <EOL> test_values [ row . key_string ] = self . randomizeRow ( row ) <EOL> ds . append ( self . reflector . insertRow ( row ) ) <EOL> return defer . gatherResults ( ds ) <EOL> def _loadRowsBack ( _ ) : <EOL> d = self . reflector . loadObjectsFrom ( "<STR_LIT>" ) <EOL> return d . addCallback ( self . gotData ) <EOL> def _checkRowsBack ( _ ) : <EOL> self . failUnless ( len ( self . data ) == self . num_iterations + <NUM_LIT:1> , <EOL> "<STR_LIT>" ) <EOL> for row in self . data : <EOL> self . failUnless ( rowMatches ( row , test_values [ row . key_string ] ) , <EOL> "<STR_LIT>" % row . key_string ) <EOL> def _changeRows ( _ ) : <EOL> ds = [ ] <EOL> for row in self . data : <EOL> test_values [ row . key_string ] = self . randomizeRow ( row ) <EOL> ds . append ( self . reflector . updateRow ( row ) ) <EOL> d = defer . gatherResults ( ds ) <EOL> return d . addCallback ( _cbChangeRows ) <EOL> def _cbChangeRows ( _ ) : <EOL> self . data = None <EOL> def _deleteRows ( _ ) : <EOL> ds = [ ] <EOL> for row in self . data : <EOL> ds . append ( self . reflector . deleteRow ( row ) ) <EOL> d = defer . gatherResults ( ds ) <EOL> return d . addCallback ( _cbChangeRows ) <EOL> def _checkRowsDeleted ( _ ) : <EOL> self . failUnless ( len ( self . data ) == <NUM_LIT:0> , "<STR_LIT>" ) <EOL> d . addCallback ( _loadObjects ) <EOL> d . addCallback ( _checkLoadObjects ) <EOL> d . addCallback ( _loadObjects ) <EOL> d . addCallback ( _checkLoadObjects2 ) <EOL> d . addCallback ( _changeParent ) <EOL> d . addCallback ( _loadBack ) <EOL> d . addCallback ( _checkLoadBack ) <EOL> d . addCallback ( _saveMoreTestRows ) <EOL> d . addCallback ( _loadRowsBack ) <EOL> d . addCallback ( _checkRowsBack ) <EOL> d . addCallback ( _changeRows ) <EOL> d . addCallback ( _loadRowsBack ) <EOL> d . addCallback ( _checkRowsBack ) <EOL> d . addCallback ( _deleteRows ) <EOL> d . addCallback ( _loadRowsBack ) <EOL> d . addCallback ( _checkRowsDeleted ) <EOL> return d <EOL> def test_saveAndDelete ( self ) : <EOL> """<STR_LIT>""" <EOL> row = TestRow ( ) <EOL> row . assignKeyAttr ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> values = self . randomizeRow ( row ) <EOL> d = self . reflector . insertRow ( row ) <EOL> def _deleteRow ( _ ) : <EOL> return self . reflector . deleteRow ( row ) <EOL> d . addCallback ( _deleteRow ) <EOL> return d <EOL> test_saveAndDelete . suppress = [ rowObjectSuppression , reflectorSuppression ] <EOL> def gotData ( self , data ) : <EOL> self . data = data <EOL> ReflectorTestBase . timeout = <NUM_LIT> <EOL> class SQLReflectorTestBase ( ReflectorTestBase ) : <EOL> """<STR_LIT>""" <EOL> def createReflector ( self ) : <EOL> self . startDB ( ) <EOL> self . dbpool = self . makePool ( ) <EOL> self . dbpool . start ( ) <EOL> if self . can_clear : <EOL> d = self . dbpool . runOperation ( '<STR_LIT>' ) <EOL> d . addCallback ( lambda _ : <EOL> self . dbpool . runOperation ( '<STR_LIT>' ) ) <EOL> d . addErrback ( lambda _ : None ) <EOL> else : <EOL> d = defer . succeed ( None ) <EOL> d . addCallback ( lambda _ : self . dbpool . runOperation ( main_table_schema ) ) <EOL> d . addCallback ( lambda _ : self . dbpool . runOperation ( child_table_schema ) ) <EOL> reflectorClass = self . escape_slashes and SQLReflector or NoSlashSQLReflector <EOL> d . addCallback ( lambda _ : <EOL> reflectorClass ( self . dbpool , [ TestRow , ChildRow ] ) ) <EOL> return d <EOL> def destroyReflector ( self ) : <EOL> d = self . dbpool . runOperation ( '<STR_LIT>' ) <EOL> d . addCallback ( lambda _ : <EOL> self . dbpool . runOperation ( '<STR_LIT>' ) ) <EOL> def close ( _ ) : <EOL> self . dbpool . close ( ) <EOL> self . stopDB ( ) <EOL> d . addCallback ( close ) <EOL> return d <EOL> class DeprecationTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_rowDeprecation ( self ) : <EOL> """<STR_LIT>""" <EOL> def wrapper ( ) : <EOL> return TestRow ( ) <EOL> self . assertWarns ( DeprecationWarning , <EOL> "<STR_LIT>" , <EOL> __file__ , <EOL> wrapper ) <EOL> def test_reflectorDeprecation ( self ) : <EOL> """<STR_LIT>""" <EOL> def wrapper ( ) : <EOL> return SQLReflector ( None , ( ) ) <EOL> from twisted . enterprise import sqlreflector <EOL> self . assertWarns ( DeprecationWarning , <EOL> "<STR_LIT>" , <EOL> sqlreflector . __file__ , <EOL> wrapper ) <EOL> makeSQLTests ( SQLReflectorTestBase , '<STR_LIT>' , globals ( ) ) <EOL> class NoSlashSQLReflector ( SQLReflector ) : <EOL> """<STR_LIT>""" <EOL> def escape_string ( self , text ) : <EOL> return text . replace ( "<STR_LIT:'>" , "<STR_LIT>" ) </s>
