Fix race condition in event assertion

tests/integration/reactor/test_reactor.py

@@ -35,4 +35,4 @@ def test_ping_reaction(self):
 
         e.fire_event({'a': 'b'}, '/test_event')
 
-        self.assertMinionEventReceived({'a': 'b'})
+        self.assertMinionEventReceived({'a': 'b'}, timeout=30)

tests/support/mixins.py

@@ -47,6 +47,7 @@
 # Import 3rd-party libs
 from salt.ext import six
 from salt.ext.six.moves import zip  # pylint: disable=import-error,redefined-builtin
+from salt.ext.six.moves.queue import Empty  # pylint: disable=import-error,no-name-in-module
 
 log = logging.getLogger(__name__)
 
@@ -687,19 +688,22 @@ def assertMinionEventFired(self, tag):
         #TODO
         raise salt.exceptions.NotImplemented('assertMinionEventFired() not implemented')
 
-    def assertMinionEventReceived(self, desired_event):
-        queue_wait = 5  # 2.5s
-        while self.q.empty():
-            time.sleep(0.5)  # Wait for events to be pushed into the queue
-            queue_wait -= 1
-            if queue_wait <= 0:
-                raise AssertionError('Queue wait timer expired')
-        while not self.q.empty():  # This is not thread-safe and may be inaccurate
-            event = self.q.get()
+    def assertMinionEventReceived(self, desired_event, timeout=5, sleep_time=0.5):
+        start = time.time()
+        while True:
+            try:
+                event = self.q.get(False)
+            except Empty:
+                time.sleep(sleep_time)
+                if start - time.time() >= timeout:
+                    break
+                continue
             if isinstance(event, dict):
                 event.pop('_stamp')
             if desired_event == event:
                 self.fetch_proc.terminate()
                 return True
+            if start - time.time() >= timeout:
+                break
         self.fetch_proc.terminate()
         raise AssertionError('Event {0} was not received by minion'.format(desired_event))