<s> import StringIO <EOL> import unittest <EOL> import weakref <EOL> import sys <EOL> import gc <EOL> import IECore <EOL> import Gaffer <EOL> import GafferTest <EOL> class SignalsTest ( GafferTest . TestCase ) : <EOL> def test ( self ) : <EOL> def f ( a ) : <EOL> return int ( a ) <EOL> s = Gaffer . Signal1 ( ) <EOL> c = s . connect ( f ) <EOL> self . assertEqual ( c . connected ( ) , True ) <EOL> self . assertEqual ( c . blocked ( ) , False ) <EOL> self . assertEqual ( s ( <NUM_LIT> ) , <NUM_LIT:5> ) <EOL> c . block ( ) <EOL> self . assertEqual ( c . blocked ( ) , True ) <EOL> c . unblock ( ) <EOL> self . assertEqual ( c . blocked ( ) , False ) <EOL> c . disconnect ( ) <EOL> self . assertEqual ( c . connected ( ) , False ) <EOL> def testDisconnectWhenSignalDies ( self ) : <EOL> def f ( a ) : <EOL> return int ( a ) <EOL> s = Gaffer . Signal1 ( ) <EOL> c = s . connect ( f ) <EOL> self . assert_ ( c . connected ( ) ) <EOL> del s <EOL> self . assert_ ( not c . connected ( ) ) <EOL> def test2 ( self ) : <EOL> def f ( a , b ) : <EOL> return a * b <EOL> s = Gaffer . Signal2 ( ) <EOL> c = s . connect ( f ) <EOL> self . assertEqual ( s ( <NUM_LIT> , <NUM_LIT> ) , <NUM_LIT> ) <EOL> def testCircularRef ( self ) : <EOL> def default ( a ) : <EOL> return - <NUM_LIT:1> <EOL> class A ( IECore . V3f ) : <EOL> def __init__ ( self ) : <EOL> IECore . V3f . __init__ ( self ) <EOL> self . signal = Gaffer . Signal1 ( ) <EOL> def f ( self , n ) : <EOL> return int ( n * <NUM_LIT:2> ) <EOL> a1 = A ( ) <EOL> a2 = A ( ) <EOL> defaultConnection = a2 . signal . connect ( default ) <EOL> self . assertEqual ( a2 . signal ( <NUM_LIT:2> ) , - <NUM_LIT:1> ) <EOL> a1 . c = a2 . signal . connect ( Gaffer . WeakMethod ( a1 . f ) ) <EOL> self . assertEqual ( a2 . signal ( <NUM_LIT:2> ) , <NUM_LIT:4> ) <EOL> a2 . c = a1 . signal . connect ( Gaffer . WeakMethod ( a2 . f ) ) <EOL> self . assert_ ( a2 . c . connected ( ) ) <EOL> self . assertEqual ( a1 . signal ( <NUM_LIT:2> ) , <NUM_LIT:4> ) <EOL> del a1 <EOL> self . assertEqual ( a2 . signal ( <NUM_LIT:2> ) , - <NUM_LIT:1> ) <EOL> self . assert_ ( not a2 . c . connected ( ) ) <EOL> def testDeletionOfConnectionDisconnects ( self ) : <EOL> def default ( a ) : <EOL> return - <NUM_LIT:1> <EOL> def f ( a ) : <EOL> return int ( a * <NUM_LIT:10> ) <EOL> s = Gaffer . Signal1 ( ) <EOL> dc = s . connect ( default ) <EOL> self . assertEqual ( s ( <NUM_LIT:1> ) , - <NUM_LIT:1> ) <EOL> c = s . connect ( f ) <EOL> self . assertEqual ( s ( <NUM_LIT:1> ) , <NUM_LIT:10> ) <EOL> del c <EOL> self . assertEqual ( s ( <NUM_LIT:1> ) , - <NUM_LIT:1> ) <EOL> def testMany ( self ) : <EOL> class S ( IECore . V3f ) : <EOL> instances = <NUM_LIT:0> <EOL> def __init__ ( self , parent ) : <EOL> IECore . V3f . __init__ ( self ) <EOL> S . instances += <NUM_LIT:1> <EOL> self . children = [ ] <EOL> self . numConnections = <NUM_LIT:0> <EOL> self . signal = Gaffer . Signal1 ( ) <EOL> if parent : <EOL> self . c = parent . signal . connect ( self . f ) <EOL> parent . numConnections += <NUM_LIT:1> <EOL> parent . children . append ( self ) <EOL> def f ( self , a ) : <EOL> r = <NUM_LIT:1> <EOL> if self . numConnections != <NUM_LIT:0> : <EOL> r += self . signal ( a ) <EOL> return r <EOL> def build ( parent , depth = <NUM_LIT:0> ) : <EOL> if ( depth > <NUM_LIT:15> ) : <EOL> return <EOL> else : <EOL> s1 = S ( parent ) <EOL> s2 = S ( parent ) <EOL> build ( s1 , depth + <NUM_LIT:1> ) <EOL> build ( s2 , depth + <NUM_LIT:1> ) <EOL> s = S ( None ) <EOL> build ( s ) <EOL> s . signal ( <NUM_LIT:1> ) <EOL> def testExceptionRefCounting ( self ) : <EOL> class T ( object ) : <EOL> def __init__ ( self , s ) : <EOL> self . connection = s . memberAddedSignal ( ) . connect ( Gaffer . WeakMethod ( self . callback ) ) <EOL> def callback ( self , s , n ) : <EOL> raise Exception <EOL> s = Gaffer . StandardSet ( ) <EOL> t = T ( s ) <EOL> w = weakref . ref ( t ) <EOL> realStdErr = sys . stderr <EOL> sio = StringIO . StringIO ( ) <EOL> try : <EOL> sys . stderr = sio <EOL> s . add ( Gaffer . Node ( ) ) <EOL> finally : <EOL> sys . stderr = realStdErr <EOL> del t <EOL> self . assert_ ( w ( ) is None ) <EOL> self . assert_ ( "<STR_LIT>" in sio . getvalue ( ) ) <EOL> def test0Arity ( self ) : <EOL> def one ( ) : <EOL> return <NUM_LIT:1> <EOL> s = Gaffer . Signal0 ( ) <EOL> c = s . connect ( one ) <EOL> self . assertEqual ( s ( ) , <NUM_LIT:1> ) <EOL> def testGenericPythonSignals ( self ) : <EOL> def one ( ) : <EOL> return "<STR_LIT>" <EOL> def two ( ) : <EOL> return "<STR_LIT>" <EOL> s = Gaffer . Signal0 ( ) <EOL> c1 = s . connect ( one ) <EOL> c2 = s . connect ( two ) <EOL> self . assertEqual ( s ( ) , "<STR_LIT>" ) <EOL> def testGenericPythonSignalsWithCombiner ( self ) : <EOL> def myCombiner ( slotResults ) : <EOL> l = [ ] <EOL> for r in slotResults : <EOL> l . append ( r ) <EOL> return l <EOL> def add ( a , b ) : <EOL> return a + b <EOL> def mult ( a , b ) : <EOL> return a * b <EOL> s = Gaffer . Signal2 ( myCombiner ) <EOL> addConnection = s . connect ( add ) <EOL> multConnection = s . connect ( mult ) <EOL> self . assertEqual ( s ( <NUM_LIT:2> , <NUM_LIT:4> ) , [ <NUM_LIT:6> , <NUM_LIT:8> ] ) <EOL> def testPythonResultCombinersCanSkipSlots ( self ) : <EOL> def myCombiner ( slotResults ) : <EOL> for r in slotResults : <EOL> if r : <EOL> return r <EOL> return False <EOL> def slot1 ( ) : <EOL> self . numCalls += <NUM_LIT:1> <EOL> return True <EOL> def slot2 ( ) : <EOL> self . numCalls += <NUM_LIT:1> <EOL> return False <EOL> s = Gaffer . Signal0 ( myCombiner ) <EOL> c1 = s . connect ( slot1 ) <EOL> c2 = s . connect ( slot2 ) <EOL> self . numCalls = <NUM_LIT:0> <EOL> self . assertEqual ( s ( ) , True ) <EOL> self . assertEqual ( self . numCalls , <NUM_LIT:1> ) <EOL> def testGroupingAndOrdering ( self ) : <EOL> values = [ ] <EOL> def f ( value ) : <EOL> values . append ( value ) <EOL> s = Gaffer . Signal0 ( ) <EOL> c1 = s . connect ( IECore . curry ( f , "<STR_LIT>" ) ) <EOL> c2 = s . connect ( IECore . curry ( f , "<STR_LIT>" ) ) <EOL> s ( ) <EOL> self . assertEqual ( values , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> del values [ : ] <EOL> c1 = s . connect ( <NUM_LIT:1> , IECore . curry ( f , "<STR_LIT>" ) ) <EOL> c2 = s . connect ( <NUM_LIT:0> , IECore . curry ( f , "<STR_LIT>" ) ) <EOL> s ( ) <EOL> self . assertEqual ( values , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> del values [ : ] <EOL> c1 = s . connect ( IECore . curry ( f , "<STR_LIT>" ) ) <EOL> c2 = s . connect ( <NUM_LIT:0> , IECore . curry ( f , "<STR_LIT>" ) ) <EOL> s ( ) <EOL> self . assertEqual ( values , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def testSlotQueries ( self ) : <EOL> def f ( ) : <EOL> pass <EOL> s = Gaffer . Signal0 ( ) <EOL> self . assertTrue ( s . empty ( ) ) <EOL> self . assertEqual ( s . num_slots ( ) , <NUM_LIT:0> ) <EOL> c = s . connect ( f ) <EOL> self . assertFalse ( s . empty ( ) ) <EOL> self . assertEqual ( s . num_slots ( ) , <NUM_LIT:1> ) <EOL> del c <EOL> self . assertTrue ( s . empty ( ) ) <EOL> self . assertEqual ( s . num_slots ( ) , <NUM_LIT:0> ) <EOL> def testNonScopedConnection ( self ) : <EOL> self . numCalls = <NUM_LIT:0> <EOL> def f ( ) : <EOL> self . numCalls += <NUM_LIT:1> <EOL> s = Gaffer . Signal0 ( ) <EOL> c = s . connect ( f , scoped = False ) <EOL> self . assertEqual ( self . numCalls , <NUM_LIT:0> ) <EOL> s ( ) <EOL> self . assertEqual ( self . numCalls , <NUM_LIT:1> ) <EOL> c . block ( True ) <EOL> s ( ) <EOL> self . assertEqual ( self . numCalls , <NUM_LIT:1> ) <EOL> c . block ( False ) <EOL> s ( ) <EOL> self . assertEqual ( self . numCalls , <NUM_LIT:2> ) <EOL> w = weakref . ref ( f ) <EOL> del f <EOL> self . assertTrue ( w ( ) is not None ) <EOL> s ( ) <EOL> self . assertEqual ( self . numCalls , <NUM_LIT:3> ) <EOL> del s <EOL> self . assertTrue ( w ( ) is None ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
