<s> import gc <EOL> import sys <EOL> import time <EOL> import unittest <EOL> import weakref <EOL> from types import TracebackType <EOL> from django . dispatch import Signal , receiver <EOL> if sys . platform . startswith ( '<STR_LIT>' ) : <EOL> def garbage_collect ( ) : <EOL> gc . collect ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> elif hasattr ( sys , "<STR_LIT>" ) : <EOL> def garbage_collect ( ) : <EOL> gc . collect ( ) <EOL> gc . collect ( ) <EOL> else : <EOL> def garbage_collect ( ) : <EOL> gc . collect ( ) <EOL> def receiver_1_arg ( val , ** kwargs ) : <EOL> return val <EOL> class Callable ( object ) : <EOL> def __call__ ( self , val , ** kwargs ) : <EOL> return val <EOL> def a ( self , val , ** kwargs ) : <EOL> return val <EOL> a_signal = Signal ( providing_args = [ "<STR_LIT>" ] ) <EOL> b_signal = Signal ( providing_args = [ "<STR_LIT>" ] ) <EOL> c_signal = Signal ( providing_args = [ "<STR_LIT>" ] ) <EOL> d_signal = Signal ( providing_args = [ "<STR_LIT>" ] , use_caching = True ) <EOL> class DispatcherTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def assertTestIsClean ( self , signal ) : <EOL> """<STR_LIT>""" <EOL> self . assertFalse ( signal . has_listeners ( ) ) <EOL> self . assertEqual ( signal . receivers , [ ] ) <EOL> def test_exact ( self ) : <EOL> a_signal . connect ( receiver_1_arg , sender = self ) <EOL> expected = [ ( receiver_1_arg , "<STR_LIT:test>" ) ] <EOL> result = a_signal . send ( sender = self , val = "<STR_LIT:test>" ) <EOL> self . assertEqual ( result , expected ) <EOL> a_signal . disconnect ( receiver_1_arg , sender = self ) <EOL> self . assertTestIsClean ( a_signal ) <EOL> def test_ignored_sender ( self ) : <EOL> a_signal . connect ( receiver_1_arg ) <EOL> expected = [ ( receiver_1_arg , "<STR_LIT:test>" ) ] <EOL> result = a_signal . send ( sender = self , val = "<STR_LIT:test>" ) <EOL> self . assertEqual ( result , expected ) <EOL> a_signal . disconnect ( receiver_1_arg ) <EOL> self . assertTestIsClean ( a_signal ) <EOL> def test_garbage_collected ( self ) : <EOL> a = Callable ( ) <EOL> a_signal . connect ( a . a , sender = self ) <EOL> expected = [ ] <EOL> del a <EOL> garbage_collect ( ) <EOL> result = a_signal . send ( sender = self , val = "<STR_LIT:test>" ) <EOL> self . assertEqual ( result , expected ) <EOL> self . assertTestIsClean ( a_signal ) <EOL> def test_cached_garbaged_collected ( self ) : <EOL> """<STR_LIT>""" <EOL> class sender : <EOL> pass <EOL> wref = weakref . ref ( sender ) <EOL> d_signal . connect ( receiver_1_arg ) <EOL> d_signal . send ( sender , val = '<STR_LIT>' ) <EOL> del sender <EOL> garbage_collect ( ) <EOL> try : <EOL> self . assertIsNone ( wref ( ) ) <EOL> finally : <EOL> d_signal . disconnect ( receiver_1_arg ) <EOL> def test_multiple_registration ( self ) : <EOL> a = Callable ( ) <EOL> a_signal . connect ( a ) <EOL> a_signal . connect ( a ) <EOL> a_signal . connect ( a ) <EOL> a_signal . connect ( a ) <EOL> a_signal . connect ( a ) <EOL> a_signal . connect ( a ) <EOL> result = a_signal . send ( sender = self , val = "<STR_LIT:test>" ) <EOL> self . assertEqual ( len ( result ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( a_signal . receivers ) , <NUM_LIT:1> ) <EOL> del a <EOL> del result <EOL> garbage_collect ( ) <EOL> self . assertTestIsClean ( a_signal ) <EOL> def test_uid_registration ( self ) : <EOL> def uid_based_receiver_1 ( ** kwargs ) : <EOL> pass <EOL> def uid_based_receiver_2 ( ** kwargs ) : <EOL> pass <EOL> a_signal . connect ( uid_based_receiver_1 , dispatch_uid = "<STR_LIT>" ) <EOL> a_signal . connect ( uid_based_receiver_2 , dispatch_uid = "<STR_LIT>" ) <EOL> self . assertEqual ( len ( a_signal . receivers ) , <NUM_LIT:1> ) <EOL> a_signal . disconnect ( dispatch_uid = "<STR_LIT>" ) <EOL> self . assertTestIsClean ( a_signal ) <EOL> def test_robust ( self ) : <EOL> """<STR_LIT>""" <EOL> def fails ( val , ** kwargs ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> a_signal . connect ( fails ) <EOL> result = a_signal . send_robust ( sender = self , val = "<STR_LIT:test>" ) <EOL> err = result [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> self . assertIsInstance ( err , ValueError ) <EOL> self . assertEqual ( err . args , ( '<STR_LIT>' , ) ) <EOL> self . assertTrue ( hasattr ( err , '<STR_LIT>' ) ) <EOL> self . assertIsInstance ( err . __traceback__ , TracebackType ) <EOL> a_signal . disconnect ( fails ) <EOL> self . assertTestIsClean ( a_signal ) <EOL> def test_disconnection ( self ) : <EOL> receiver_1 = Callable ( ) <EOL> receiver_2 = Callable ( ) <EOL> receiver_3 = Callable ( ) <EOL> a_signal . connect ( receiver_1 ) <EOL> a_signal . connect ( receiver_2 ) <EOL> a_signal . connect ( receiver_3 ) <EOL> a_signal . disconnect ( receiver_1 ) <EOL> del receiver_2 <EOL> garbage_collect ( ) <EOL> a_signal . disconnect ( receiver_3 ) <EOL> self . assertTestIsClean ( a_signal ) <EOL> def test_values_returned_by_disconnection ( self ) : <EOL> receiver_1 = Callable ( ) <EOL> receiver_2 = Callable ( ) <EOL> a_signal . connect ( receiver_1 ) <EOL> receiver_1_disconnected = a_signal . disconnect ( receiver_1 ) <EOL> receiver_2_disconnected = a_signal . disconnect ( receiver_2 ) <EOL> self . assertTrue ( receiver_1_disconnected ) <EOL> self . assertFalse ( receiver_2_disconnected ) <EOL> self . assertTestIsClean ( a_signal ) <EOL> def test_has_listeners ( self ) : <EOL> self . assertFalse ( a_signal . has_listeners ( ) ) <EOL> self . assertFalse ( a_signal . has_listeners ( sender = object ( ) ) ) <EOL> receiver_1 = Callable ( ) <EOL> a_signal . connect ( receiver_1 ) <EOL> self . assertTrue ( a_signal . has_listeners ( ) ) <EOL> self . assertTrue ( a_signal . has_listeners ( sender = object ( ) ) ) <EOL> a_signal . disconnect ( receiver_1 ) <EOL> self . assertFalse ( a_signal . has_listeners ( ) ) <EOL> self . assertFalse ( a_signal . has_listeners ( sender = object ( ) ) ) <EOL> class ReceiverTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_receiver_single_signal ( self ) : <EOL> @ receiver ( a_signal ) <EOL> def f ( val , ** kwargs ) : <EOL> self . state = val <EOL> self . state = False <EOL> a_signal . send ( sender = self , val = True ) <EOL> self . assertTrue ( self . state ) <EOL> def test_receiver_signal_list ( self ) : <EOL> @ receiver ( [ a_signal , b_signal , c_signal ] ) <EOL> def f ( val , ** kwargs ) : <EOL> self . state . append ( val ) <EOL> self . state = [ ] <EOL> a_signal . send ( sender = self , val = '<STR_LIT:a>' ) <EOL> c_signal . send ( sender = self , val = '<STR_LIT:c>' ) <EOL> b_signal . send ( sender = self , val = '<STR_LIT:b>' ) <EOL> self . assertIn ( '<STR_LIT:a>' , self . state ) <EOL> self . assertIn ( '<STR_LIT:b>' , self . state ) <EOL> self . assertIn ( '<STR_LIT:c>' , self . state ) </s>
