<s> import unittest <EOL> from nose . tools import assert_equals , assert_false , assert_true <EOL> from robotide . publish import RideMessage , RideLog , PUBLISHER <EOL> from robotide . pluginapi import Plugin <EOL> class RideTestMessage ( RideMessage ) : <EOL> pass <EOL> class RideMessageWithData ( RideMessage ) : <EOL> data = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> class TestSubscribingToEvents ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . plugin = SubscribingPlugin ( ) <EOL> def tearDown ( self ) : <EOL> self . plugin . disable ( ) <EOL> def test_subscribing_with_class ( self ) : <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . class_handler_topic , '<STR_LIT>' ) <EOL> def test_subscribing_with_string ( self ) : <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . string_handler_topic , '<STR_LIT>' ) <EOL> def test_subscribing_with_string_is_case_insensitive ( self ) : <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . case_insensitive_string_handler_topic , <EOL> '<STR_LIT>' ) <EOL> def test_event_with_data ( self ) : <EOL> RideMessageWithData ( data_item = '<STR_LIT>' , more_data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) . publish ( ) <EOL> assert_equals ( self . plugin . record [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> assert_equals ( self . plugin . record [ '<STR_LIT>' ] , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> def test_subscribing_multiple_times ( self ) : <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . count , <NUM_LIT:5> ) <EOL> def test_subscribing_to_multiple_topics ( self ) : <EOL> RideMessageWithData ( data_item = '<STR_LIT>' , more_data = { } ) . publish ( ) <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . multi_events , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_subscribing_to_hierarchy ( self ) : <EOL> RideTestMessage ( ) . publish ( ) <EOL> RideMessageWithData ( data_item = None , more_data = [ ] ) . publish ( ) <EOL> assert_equals ( self . plugin . hierarchy_events , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> class TestUnsubscribingFromEvents ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . plugin = SubscribingPlugin ( ) <EOL> self . _unsubscribe_all = True <EOL> def tearDown ( self ) : <EOL> if self . _unsubscribe_all : <EOL> self . plugin . unsubscribe_all ( ) <EOL> def test_unsubscribe_with_class ( self ) : <EOL> listener_count = len ( PUBLISHER . _listeners [ self . plugin ] ) <EOL> self . plugin . unsubscribe ( self . plugin . OnTestEventClass , RideTestMessage ) <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . class_handler_topic , None ) <EOL> assert_equals ( <EOL> len ( PUBLISHER . _listeners [ self . plugin ] ) , listener_count - <NUM_LIT:1> ) <EOL> def test_unsubscribe_with_string ( self ) : <EOL> self . plugin . unsubscribe ( self . plugin . OnTestEventString , '<STR_LIT>' ) <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . string_handler_topic , None ) <EOL> def test_unsubscribe_with_string_is_case_insensitive ( self ) : <EOL> self . plugin . unsubscribe ( <EOL> self . plugin . OnTestEventStringWrongCase , '<STR_LIT>' ) <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . case_insensitive_string_handler_topic , None ) <EOL> def test_unsubscribing_multiple_times_subscribed_once ( self ) : <EOL> self . plugin . unsubscribe ( self . plugin . counting_handler , RideTestMessage ) <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . count , <NUM_LIT:4> ) <EOL> def test_unsubscribing_multiple_times_subscribed_all ( self ) : <EOL> for _ in range ( <NUM_LIT:5> ) : <EOL> self . plugin . unsubscribe ( <EOL> self . plugin . counting_handler , RideTestMessage ) <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . count , <NUM_LIT:0> ) <EOL> def test_unsubscribing_from_hierarchy ( self ) : <EOL> self . plugin . unsubscribe ( self . plugin . hierarchical_listener , RideMessage ) <EOL> RideTestMessage ( ) . publish ( ) <EOL> RideMessageWithData ( data_item = '<STR_LIT>' , more_data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) . publish ( ) <EOL> assert_equals ( self . plugin . hierarchy_events , [ ] ) <EOL> def test_unsubscribing_from_one_of_the_multiple_topics ( self ) : <EOL> self . plugin . unsubscribe ( self . plugin . multiple_events_listening_handler , <EOL> RideMessageWithData ) <EOL> RideMessageWithData ( data_item = '<STR_LIT:data>' , more_data = '<STR_LIT>' ) . publish ( ) <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_equals ( self . plugin . multi_events , [ '<STR_LIT>' ] ) <EOL> def test_unsubscribing_from_multiple_topics ( self ) : <EOL> self . plugin . unsubscribe ( self . plugin . multiple_events_listening_handler , <EOL> '<STR_LIT>' , RideMessageWithData ) <EOL> RideTestMessage ( ) . publish ( ) <EOL> RideMessageWithData ( data_item = '<STR_LIT:data>' , more_data = '<STR_LIT>' ) . publish ( ) <EOL> assert_equals ( self . plugin . multi_events , [ ] ) <EOL> def test_unsubscribing_from_not_subscribed_event_does_not_fail ( self ) : <EOL> self . plugin . unsubscribe ( self . plugin . OnTestEventClass , '<STR_LIT>' ) <EOL> def test_unsubscribe_all ( self ) : <EOL> self . plugin . unsubscribe_all ( ) <EOL> self . _unsubscribe_all = False <EOL> RideTestMessage ( ) . publish ( ) <EOL> RideMessageWithData ( data_item = '<STR_LIT>' , more_data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) . publish ( ) <EOL> assert_equals ( self . plugin . class_handler_topic , None ) <EOL> assert_equals ( self . plugin . string_handler_topic , None ) <EOL> assert_equals ( self . plugin . case_insensitive_string_handler_topic , None ) <EOL> assert_equals ( self . plugin . record , { } ) <EOL> assert_equals ( self . plugin . count , <NUM_LIT:0> ) <EOL> assert_equals ( self . plugin . hierarchy_events , [ ] ) <EOL> assert_false ( self . plugin in PUBLISHER . _listeners ) <EOL> class TestBrokenMessageListener ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . plugin = BrokenListenerPlugin ( ) <EOL> def tearDown ( self ) : <EOL> self . plugin . disable ( ) <EOL> def test_broken_listener ( self ) : <EOL> self . plugin . subscribe ( self . plugin . error_listener , RideLog ) <EOL> RideTestMessage ( ) . publish ( ) <EOL> assert_true ( self . plugin . error . message . startswith ( <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' + self . plugin . error . message ) <EOL> assert_equals ( self . plugin . error . topic , '<STR_LIT>' ) <EOL> assert_equals ( self . plugin . error . level , '<STR_LIT>' ) <EOL> def test_broken_error_listener_does_not_cause_infinite_recusrion ( self ) : <EOL> self . plugin . subscribe ( self . plugin . broken_listener , RideLog ) <EOL> class BrokenListenerPlugin ( Plugin ) : <EOL> def __init__ ( self ) : <EOL> self . subscribe ( self . broken_listener , RideTestMessage ) <EOL> def disable ( self ) : <EOL> self . unsubscribe_all ( ) <EOL> def broken_listener ( self , message ) : <EOL> raise RuntimeError ( message . topic ) <EOL> def error_listener ( self , message ) : <EOL> self . error = message <EOL> class SubscribingPlugin ( Plugin ) : <EOL> def __init__ ( self ) : <EOL> self . _reset_recorders ( ) <EOL> self . _subscribe_to_events ( ) <EOL> def disable ( self ) : <EOL> self . unsubscribe_all ( ) <EOL> def _reset_recorders ( self ) : <EOL> self . record = { } <EOL> self . count = <NUM_LIT:0> <EOL> self . hierarchy_events = [ ] <EOL> self . multi_events = [ ] <EOL> self . class_handler_topic = self . string_handler_topic = self . case_insensitive_string_handler_topic = None <EOL> def _subscribe_to_events ( self ) : <EOL> self . subscribe ( self . OnTestEventClass , RideTestMessage ) <EOL> self . subscribe ( self . OnTestEventString , '<STR_LIT>' ) <EOL> self . subscribe ( self . OnTestEventStringWrongCase , '<STR_LIT>' ) <EOL> self . subscribe ( self . OnTestEventWithData , RideMessageWithData ) <EOL> for _ in range ( <NUM_LIT:5> ) : <EOL> self . subscribe ( self . counting_handler , RideTestMessage ) <EOL> self . subscribe ( self . hierarchical_listener , RideMessage ) <EOL> self . subscribe ( self . multiple_events_listening_handler , RideTestMessage , <EOL> RideMessageWithData ) <EOL> def OnTestEventClass ( self , event ) : <EOL> self . class_handler_topic = event . topic <EOL> def OnTestEventString ( self , event ) : <EOL> self . string_handler_topic = event . topic <EOL> def OnTestEventStringWrongCase ( self , event ) : <EOL> self . case_insensitive_string_handler_topic = event . topic <EOL> def OnTestEventWithData ( self , event ) : <EOL> self . record [ '<STR_LIT>' ] = event . data_item <EOL> self . record [ '<STR_LIT>' ] = event . more_data <EOL> def counting_handler ( self , event ) : <EOL> self . count += <NUM_LIT:1> <EOL> def hierarchical_listener ( self , event ) : <EOL> self . hierarchy_events . append ( event . topic ) <EOL> def multiple_events_listening_handler ( self , event ) : <EOL> self . multi_events . append ( event . topic ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
