def testEventNotification(self):
    NoClientListener.received_events = []
    worker = test_lib.MockWorker(token=self.token)

    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")

    # Not allowed to publish a message from a client..
    flow.Events.PublishEvent("TestEvent", event, token=self.token)
    worker.Simulate()

    self.assertEqual(NoClientListener.received_events, [])

    event.source = "Source"

    # First make the message unauthenticated.
    event.auth_state = rdf_flows.GrrMessage.AuthorizationState.UNAUTHENTICATED

    # Publish the event.
    flow.Events.PublishEvent("TestEvent", event, token=self.token)
    worker.Simulate()

    # This should not work - the unauthenticated message is dropped.
    self.assertEqual(NoClientListener.received_events, [])

    # Now make the message authenticated.
    event.auth_state = rdf_flows.GrrMessage.AuthorizationState.AUTHENTICATED

    # Publish the event.
    flow.Events.PublishEvent("TestEvent", event, token=self.token)
    worker.Simulate()

    # This should now work:
    self.assertEqual(len(NoClientListener.received_events), 1)

    # Make sure the source is correctly propagated.
    self.assertEqual(NoClientListener.received_events[0][0].source,
                     "aff4:/Source")
    self.assertEqual(NoClientListener.received_events[0][1].path, "foobar")

    NoClientListener.received_events = []
    # Now schedule ten events at the same time.
    for i in xrange(10):
      event.source = "Source%d" % i
      flow.Events.PublishEvent("TestEvent", event, token=self.token)

    worker.Simulate()

    self.assertEqual(len(NoClientListener.received_events), 10)

    # Events do not have to be delivered in order so we sort them here for
    # comparison.
    NoClientListener.received_events.sort(key=lambda x: x[0].source)
    for i in range(10):
      self.assertEqual(NoClientListener.received_events[i][0].source,
                       "aff4:/Source%d" % i)
      self.assertEqual(NoClientListener.received_events[i][1].path, "foobar")
Test that events are sent to listeners.
dataset/ETHPy150Open google/grr/grr/lib/flow_test.py/GeneralFlowsTest.testEventNotification