<s> import lxml . etree as etree <EOL> from twisted . internet import reactor <EOL> from twisted . internet import task <EOL> from twisted . trial import unittest <EOL> from twisted . test import proto_helpers <EOL> from comet . testutils import ( DUMMY_IAMALIVE , DUMMY_ACK , DUMMY_NAK , <EOL> DUMMY_AUTHENTICATE_RESPONSE , <EOL> DUMMY_AUTHENTICATE_RESPONSE_LEGACY , <EOL> DUMMY_SERVICE_IVORN , DummyEvent ) <EOL> from comet . service . broker import BCAST_TEST_INTERVAL <EOL> from comet . protocol . broadcaster import VOEventBroadcaster , VOEventBroadcasterFactory <EOL> class DummyBroadcaster ( object ) : <EOL> def __init__ ( self ) : <EOL> self . received_alive = False <EOL> self . received_event = False <EOL> def sendIAmAlive ( self ) : <EOL> self . received_alive = True <EOL> def send_event ( self , event ) : <EOL> self . received_event = True <EOL> class VOEventBroadcasterFactoryTestCaseBase ( unittest . TestCase ) : <EOL> def setUp ( self , test_interval = BCAST_TEST_INTERVAL ) : <EOL> self . factory = VOEventBroadcasterFactory ( <EOL> DUMMY_SERVICE_IVORN . decode ( ) , test_interval <EOL> ) <EOL> self . factory . alive_loop . clock = task . Clock ( ) <EOL> self . factory . test_loop . clock = task . Clock ( ) <EOL> self . factory . broadcasters . append ( DummyBroadcaster ( ) ) <EOL> self . connector = reactor . listenTCP ( <NUM_LIT:0> , self . factory ) <EOL> def tearDown ( self ) : <EOL> self . connector . stopListening ( ) <EOL> class VOEventBroadcasterFactoryTestCase ( VOEventBroadcasterFactoryTestCaseBase ) : <EOL> def test_protocol ( self ) : <EOL> self . assertEqual ( self . factory . protocol , VOEventBroadcaster ) <EOL> def test_sendIamAlive ( self ) : <EOL> self . assertEqual ( self . factory . alive_loop . running , True ) <EOL> self . factory . alive_loop . clock . advance ( self . factory . IAMALIVE_INTERVAL ) <EOL> for broadcaster in self . factory . broadcasters : <EOL> self . assertEqual ( broadcaster . received_alive , True ) <EOL> def test_sendTestEvent ( self ) : <EOL> self . assertEqual ( self . factory . test_loop . running , True ) <EOL> self . factory . test_loop . clock . advance ( BCAST_TEST_INTERVAL ) <EOL> for broadcaster in self . factory . broadcasters : <EOL> self . assertEqual ( broadcaster . received_event , True ) <EOL> class VOEventBroadcasterFactoryNoTestEventsTestCase ( VOEventBroadcasterFactoryTestCaseBase ) : <EOL> def setUp ( self ) : <EOL> VOEventBroadcasterFactoryTestCaseBase . setUp ( self , <NUM_LIT:0> ) <EOL> def test_DoNotSendTestEvent ( self ) : <EOL> self . assertEqual ( self . factory . test_loop . running , False ) <EOL> self . factory . test_loop . clock . advance ( BCAST_TEST_INTERVAL ) <EOL> for broadcaster in self . factory . broadcasters : <EOL> self . assertEqual ( broadcaster . received_event , False ) <EOL> class VOEventBroadcasterTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . factory = VOEventBroadcasterFactory ( <EOL> DUMMY_SERVICE_IVORN . decode ( ) , BCAST_TEST_INTERVAL <EOL> ) <EOL> self . factory . alive_loop . clock = task . Clock ( ) <EOL> self . factory . test_loop . clock = task . Clock ( ) <EOL> self . connector = reactor . listenTCP ( <NUM_LIT:0> , self . factory ) <EOL> self . proto = self . factory . buildProtocol ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) ) <EOL> self . tr = proto_helpers . StringTransportWithDisconnection ( ) <EOL> self . proto . makeConnection ( self . tr ) <EOL> self . tr . protocol = self . proto <EOL> def tearDown ( self ) : <EOL> self . connector . stopListening ( ) <EOL> def test_register_broadcaster ( self ) : <EOL> self . assertIn ( self . proto , self . factory . broadcasters ) <EOL> def test_sent_authenticate ( self ) : <EOL> received_element = etree . fromstring ( self . tr . value ( ) [ <NUM_LIT:4> : ] ) <EOL> self . assertEqual ( "<STR_LIT>" , received_element . attrib [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( DUMMY_SERVICE_IVORN . decode ( ) , received_element . find ( '<STR_LIT>' ) . text ) <EOL> def test_sendIAmAlive ( self ) : <EOL> self . tr . clear ( ) <EOL> init_alive_count = self . proto . alive_count <EOL> self . factory . alive_loop . clock . advance ( self . factory . IAMALIVE_INTERVAL ) <EOL> received_element = etree . fromstring ( self . tr . value ( ) [ <NUM_LIT:4> : ] ) <EOL> self . assertEqual ( "<STR_LIT>" , received_element . attrib [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( DUMMY_SERVICE_IVORN . decode ( ) , received_element . find ( '<STR_LIT>' ) . text ) <EOL> self . assertEqual ( self . proto . alive_count - init_alive_count , <NUM_LIT:1> ) <EOL> def test_alive_timeout ( self ) : <EOL> self . assertEqual ( self . tr . connected , True ) <EOL> for x in range ( self . proto . MAX_ALIVE_COUNT + <NUM_LIT:1> ) : <EOL> self . proto . sendIAmAlive ( ) <EOL> self . assertEqual ( self . tr . connected , False ) <EOL> def test_lack_of_ack ( self ) : <EOL> self . assertEqual ( self . tr . connected , True ) <EOL> self . proto . outstanding_ack = self . proto . MAX_OUTSTANDING_ACK + <NUM_LIT:1> <EOL> self . proto . sendIAmAlive ( ) <EOL> self . assertEqual ( self . tr . connected , False ) <EOL> def test_send_event ( self ) : <EOL> self . tr . clear ( ) <EOL> init_outstanding_ack = self . proto . outstanding_ack <EOL> self . proto . send_event ( DummyEvent ( ) ) <EOL> self . assertEqual ( self . tr . value ( ) [ <NUM_LIT:4> : ] , DummyEvent ( ) . raw_bytes ) <EOL> self . assertEqual ( self . proto . outstanding_ack - init_outstanding_ack , <NUM_LIT:1> ) <EOL> def test_send_event_with_filter_reject ( self ) : <EOL> def check_output ( result ) : <EOL> self . assertEqual ( self . tr . value ( ) , b"<STR_LIT>" ) <EOL> init_outstanding_ack = self . proto . outstanding_ack <EOL> def check_ack_increment ( result ) : <EOL> self . assertEqual ( self . proto . outstanding_ack - init_outstanding_ack , <NUM_LIT:0> ) <EOL> self . tr . clear ( ) <EOL> self . proto . filters . append ( <EOL> etree . XPath ( "<STR_LIT>" ) <EOL> ) <EOL> d = self . proto . send_event ( DummyEvent ( ) ) <EOL> d . addCallback ( check_output ) <EOL> d . addCallback ( check_ack_increment ) <EOL> return d <EOL> def test_send_event_with_filter_accept ( self ) : <EOL> def check_output ( result ) : <EOL> self . assertEqual ( self . tr . value ( ) [ <NUM_LIT:4> : ] , DummyEvent ( ) . raw_bytes ) <EOL> init_outstanding_ack = self . proto . outstanding_ack <EOL> def check_ack_increment ( result ) : <EOL> self . assertEqual ( self . proto . outstanding_ack - init_outstanding_ack , <NUM_LIT:1> ) <EOL> self . tr . clear ( ) <EOL> self . proto . filters . append ( <EOL> etree . XPath ( "<STR_LIT>" ) <EOL> ) <EOL> d = self . proto . send_event ( DummyEvent ( ) ) <EOL> d . addCallback ( check_output ) <EOL> d . addCallback ( check_ack_increment ) <EOL> return d <EOL> def test_receive_unparsable ( self ) : <EOL> self . tr . clear ( ) <EOL> unparsable = "<STR_LIT>" <EOL> self . assertRaises ( etree . ParseError , etree . fromstring , unparsable ) <EOL> self . proto . stringReceived ( unparsable ) <EOL> self . assertEqual ( self . tr . value ( ) , b"<STR_LIT>" ) <EOL> self . assertEqual ( self . tr . connected , True ) <EOL> def test_receive_incomprehensible ( self ) : <EOL> self . tr . clear ( ) <EOL> incomprehensible = b"<STR_LIT>" <EOL> etree . fromstring ( incomprehensible ) <EOL> self . proto . stringReceived ( incomprehensible ) <EOL> self . assertEqual ( self . tr . value ( ) , b"<STR_LIT>" ) <EOL> self . assertEqual ( self . tr . connected , True ) <EOL> def test_receive_iamalive ( self ) : <EOL> self . tr . clear ( ) <EOL> init_alive_count = self . proto . alive_count <EOL> self . proto . stringReceived ( DUMMY_IAMALIVE ) <EOL> self . assertEqual ( self . tr . value ( ) , b"<STR_LIT>" ) <EOL> self . assertEqual ( init_alive_count - self . proto . alive_count , <NUM_LIT:1> ) <EOL> def test_receive_ack ( self ) : <EOL> self . tr . clear ( ) <EOL> init_outstanding_ack = self . proto . outstanding_ack <EOL> self . proto . stringReceived ( DUMMY_ACK ) <EOL> self . assertEqual ( self . tr . value ( ) , b"<STR_LIT>" ) <EOL> self . assertEqual ( init_outstanding_ack - self . proto . outstanding_ack , <NUM_LIT:1> ) <EOL> def test_receive_nak ( self ) : <EOL> self . tr . clear ( ) <EOL> self . proto . stringReceived ( DUMMY_NAK ) <EOL> self . assertEqual ( self . tr . value ( ) , b"<STR_LIT>" ) <EOL> self . assertEqual ( self . tr . connected , False ) <EOL> def _test_sets_filter ( self , filter_list ) : <EOL> self . tr . clear ( ) <EOL> self . assertEqual ( len ( self . proto . filters ) , <NUM_LIT:0> ) <EOL> self . proto . stringReceived ( DUMMY_AUTHENTICATE_RESPONSE ( filter_list ) . raw_bytes ) <EOL> self . assertEqual ( self . tr . value ( ) , b"<STR_LIT>" ) <EOL> self . assertEqual ( self . tr . connected , True ) <EOL> self . assertEqual ( len ( self . proto . filters ) , len ( filter_list ) ) <EOL> def test_receive_authenticateresponse_with_filters ( self ) : <EOL> self . _test_sets_filter ( [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> ] ) <EOL> def test_receive_authenticateresponse_legacy ( self ) : <EOL> demo_filter = "<STR_LIT>" <EOL> received = ( DUMMY_AUTHENTICATE_RESPONSE_LEGACY . decode ( ) <EOL> % demo_filter ) . encode ( '<STR_LIT>' ) <EOL> self . tr . clear ( ) <EOL> self . assertEqual ( len ( self . proto . filters ) , <NUM_LIT:0> ) <EOL> self . proto . stringReceived ( received ) <EOL> self . assertEqual ( self . tr . value ( ) , b"<STR_LIT>" ) <EOL> self . assertEqual ( self . tr . connected , True ) <EOL> self . assertEqual ( len ( self . proto . filters ) , <NUM_LIT:1> ) <EOL> def test_receive_authenticateresponse_with_bad_filter ( self ) : <EOL> self . tr . clear ( ) <EOL> self . assertEqual ( len ( self . proto . filters ) , <NUM_LIT:0> ) <EOL> self . proto . stringReceived ( <EOL> DUMMY_AUTHENTICATE_RESPONSE ( [ "<STR_LIT>" ] ) . raw_bytes <EOL> ) <EOL> self . assertEqual ( self . tr . value ( ) , b"<STR_LIT>" ) <EOL> self . assertEqual ( self . tr . connected , True ) <EOL> self . assertEqual ( len ( self . proto . filters ) , <NUM_LIT:0> ) </s>
