<s> """<STR_LIT>""" <EOL> import unittest <EOL> from IPython . utils . notification import shared_center <EOL> class Observer ( object ) : <EOL> def __init__ ( self , expected_ntype , expected_sender , <EOL> center = shared_center , * args , ** kwargs ) : <EOL> super ( Observer , self ) . __init__ ( ) <EOL> self . expected_ntype = expected_ntype <EOL> self . expected_sender = expected_sender <EOL> self . expected_args = args <EOL> self . expected_kwargs = kwargs <EOL> self . recieved = False <EOL> center . add_observer ( self . callback , <EOL> self . expected_ntype , <EOL> self . expected_sender ) <EOL> def callback ( self , ntype , sender , * args , ** kwargs ) : <EOL> assert ( ntype == self . expected_ntype or <EOL> self . expected_ntype == None ) <EOL> assert ( sender == self . expected_sender or <EOL> self . expected_sender == None ) <EOL> assert ( args == self . expected_args ) <EOL> assert ( kwargs == self . expected_kwargs ) <EOL> self . recieved = True <EOL> def verify ( self ) : <EOL> assert ( self . recieved ) <EOL> def reset ( self ) : <EOL> self . recieved = False <EOL> class Notifier ( object ) : <EOL> def __init__ ( self , ntype , ** kwargs ) : <EOL> super ( Notifier , self ) . __init__ ( ) <EOL> self . ntype = ntype <EOL> self . kwargs = kwargs <EOL> def post ( self , center = shared_center ) : <EOL> center . post_notification ( self . ntype , self , <EOL> ** self . kwargs ) <EOL> class NotificationTests ( unittest . TestCase ) : <EOL> def tearDown ( self ) : <EOL> shared_center . remove_all_observers ( ) <EOL> def test_notification_delivered ( self ) : <EOL> """<STR_LIT>""" <EOL> expected_ntype = '<STR_LIT>' <EOL> sender = Notifier ( expected_ntype ) <EOL> observer = Observer ( expected_ntype , sender ) <EOL> sender . post ( ) <EOL> observer . verify ( ) <EOL> def test_type_specificity ( self ) : <EOL> """<STR_LIT>""" <EOL> expected_ntype = <NUM_LIT:1> <EOL> unexpected_ntype = "<STR_LIT>" <EOL> sender = Notifier ( expected_ntype ) <EOL> unexpected_sender = Notifier ( unexpected_ntype ) <EOL> observer = Observer ( expected_ntype , sender ) <EOL> sender . post ( ) <EOL> unexpected_sender . post ( ) <EOL> observer . verify ( ) <EOL> def test_sender_specificity ( self ) : <EOL> """<STR_LIT>""" <EOL> expected_ntype = "<STR_LIT>" <EOL> sender1 = Notifier ( expected_ntype ) <EOL> sender2 = Notifier ( expected_ntype ) <EOL> observer = Observer ( expected_ntype , sender1 ) <EOL> sender1 . post ( ) <EOL> sender2 . post ( ) <EOL> observer . verify ( ) <EOL> def test_remove_all_observers ( self ) : <EOL> """<STR_LIT>""" <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> Observer ( '<STR_LIT>' , None , center = shared_center ) <EOL> self . assert_ ( len ( shared_center . observers [ ( '<STR_LIT>' , None ) ] ) >= <NUM_LIT:10> , <EOL> "<STR_LIT>" ) <EOL> shared_center . remove_all_observers ( ) <EOL> self . assert_ ( len ( shared_center . observers ) == <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def test_any_sender ( self ) : <EOL> expected_ntype = "<STR_LIT>" <EOL> sender1 = Notifier ( expected_ntype ) <EOL> sender2 = Notifier ( expected_ntype ) <EOL> observer = Observer ( expected_ntype , None ) <EOL> sender1 . post ( ) <EOL> observer . verify ( ) <EOL> observer . reset ( ) <EOL> sender2 . post ( ) <EOL> observer . verify ( ) <EOL> def test_post_performance ( self ) : <EOL> """<STR_LIT>""" <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> Observer ( "<STR_LIT>" , None ) <EOL> o = Observer ( '<STR_LIT>' , None ) <EOL> shared_center . post_notification ( '<STR_LIT>' , self ) <EOL> o . verify ( ) </s>
