grr/server/grr_response_server/events_test.py
def testEventNotification(self):
    """Test that events are sent to listeners."""
    TestListener.received_events = []

    event = rdf_flows.GrrMessage(
        session_id=rdfvalue.SessionID(flow_name="SomeFlow"),
        name="test message",
        payload=rdf_paths.PathSpec(path="foobar", pathtype="TSK"),
        source="aff4:/C.0000000000000001",
        auth_state="AUTHENTICATED")

    events.Events.PublishEvent("TestEvent", event, token=self.token)

    # Make sure the source is correctly propagated.
    self.assertEqual(TestListener.received_events[0], event)
Test that events are sent to listeners.
Test that events are sent to listeners.
Test that events are sent to listeners .
def testEventNotification(self):
    TestListener.received_events = []
    event = rdf_flows.GrrMessage(
        session_id=rdfvalue.SessionID(flow_name="SomeFlow"),
        name="test message",
        payload=rdf_paths.PathSpec(path="foobar", pathtype="TSK"),
        source="aff4:/C.0000000000000001",
        auth_state="AUTHENTICATED")
    events.Events.PublishEvent("TestEvent", event, token=self.token)
    self.assertEqual(TestListener.received_events[0], event)
def testEventNotification ( self ) : TestListener . received_events = [ ] event = rdf_flows . GrrMessage ( session_id = rdfvalue . SessionID ( flow_name = "SomeFlow" ) , name = "test message" , payload = rdf_paths . PathSpec ( path = "foobar" , pathtype = "TSK" ) , source = "aff4:/C.0000000000000001" , auth_state = "AUTHENTICATED" ) events . Events . PublishEvent ( "TestEvent" , event , token = self . token ) self . assertEqual ( TestListener . received_events [ 0 ] , event )
Test that events are sent to listeners.
Test that events are sent to listeners .
"""Test that events are sent to listeners.""" # Make sure the source is correctly propagated.
self