Event assert (#33262)

* Initial POC

* Allow the queue to fill by waiting for the lag in IPC

* Add docs

* Add run_tests

* Add reactor to tests

doc/topics/development/tests/integration.rst

@@ -362,6 +362,29 @@ Testing salt-ssh functionality can be done using the SSHCase test class:
         self.assertEqual(cmd, 'localhost')
 
 
+Testing Event System via SaltMinionEventAssertsMixin
+----------------------------------------------------
+
+The fundamentially asynchronous nature of Salt makes testing the event system a challenge.
+The ``SaltMinionEventAssertsMixin`` provides a facility for testing that events were received
+on a minion event bus.
+
+.. code-block:: python
+
+    import integration
+
+    class TestEvent(integration.SaltEventAssertsMixin):
+        '''
+        Example test of firing an event and receiving it
+        '''
+
+        def test_event(self):
+            e = salt.utils.event.get_event('minion', sock_dir=self.minion_opts['sock_dir'], opts=self.minion_opts)
+
+            e.fire_event({'a': 'b'}, '/test_event')
+
+            self.assertMinionEventReceived({'a': 'b'})
+
 
 Syndic Example via SyndicCase
 -----------------------------

tests/integration/__init__.py

@@ -1082,6 +1082,63 @@ class AdaptedConfigurationTestCaseMixIn(object):
         return self.get_config('master')
 
 
+class SaltMinionEventAssertsMixIn(object):
+    '''
+    Asserts to verify that a given event was seen
+    '''
+
+    def __new__(cls, *args, **kwargs):
+        # We have to cross-call to re-gen a config
+        cls.q = multiprocessing.Queue()
+        cls.fetch_proc = multiprocessing.Process(target=cls._fetch, args=(cls.q,))
+        cls.fetch_proc.start()
+        return object.__new__(cls)
+
+    @staticmethod
+    def _fetch(q):
+        '''
+        Collect events and store them
+        '''
+        def _clean_queue():
+            print('Cleaning queue!')
+            while not q.empty():
+                queue_item = q.get()
+                queue_item.task_done()
+
+        atexit.register(_clean_queue)
+        a_config = AdaptedConfigurationTestCaseMixIn()
+        event = salt.utils.event.get_event('minion', sock_dir=a_config.get_config('minion')['sock_dir'], opts=a_config.get_config('minion'))
+        while True:
+            try:
+                events = event.get_event(full=False)
+            except Exception:
+                # This is broad but we'll see all kinds of issues right now
+                # if we drop the proc out from under the socket while we're reading
+                pass
+            q.put(events)
+
+    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()
+            if isinstance(event, dict):
+                event.pop('_stamp')
+            if desired_event == event:
+                self.fetch_proc.terminate()
+                return True
+        self.fetch_proc.terminate()
+        raise AssertionError('Event {0} was not received by minion'.format(desired_event))
+
+
 class SaltClientTestCaseMixIn(AdaptedConfigurationTestCaseMixIn):
 
     _salt_client_config_file_name_ = 'master'

tests/integration/reactor/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

tests/integration/reactor/reactor.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+'''
+
+    integration.reactor.reactor
+    ~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+    Test Salt's reactor system
+'''
+
+# Import Python libs
+from __future__ import absolute_import
+
+# Import Salt testing libs
+from salttesting.helpers import ensure_in_syspath
+
+ensure_in_syspath('../')
+
+import integration
+
+# Import Salt libs
+import salt.utils.event
+
+
+class ReactorTest(integration.ModuleCase, integration.SaltMinionEventAssertsMixIn):
+    '''
+    Test Salt's reactor system
+    '''
+
+    def test_ping_reaction(self):
+        '''
+        Fire an event on the master and ensure
+        that it pings the minion
+        '''
+        # Create event bus connection
+        e = salt.utils.event.get_event('minion', sock_dir=self.minion_opts['sock_dir'], opts=self.minion_opts)
+
+        e.fire_event({'a': 'b'}, '/test_event')
+
+        self.assertMinionEventReceived({'a': 'b'})
+
+if __name__ == '__main__':
+    from integration import run_tests
+    run_tests(ReactorTest)

tests/runtests.py

@@ -99,7 +99,10 @@ TEST_SUITES = {
         'path': 'integration/cloud/providers'},
     'minion':
         {'display_name': 'Minion',
-         'path': 'integration/minion'}
+         'path': 'integration/minion'},
+    'reactor':
+        {'display_name': 'Reactor',
+         'path': 'integration/reactor'},
 }
 
 
@@ -240,6 +243,13 @@ class SaltTestsuiteParser(SaltCoverageTestingParser):
             action='store_true',
             help='Run salt/renderers/*.py tests'
         )
+        self.test_selection_group.add_option(
+            '--reactor',
+            dest='reactor',
+            default=False,
+            action='store_true',
+            help='Run salt/reactor/*.py tests'
+        )
         self.test_selection_group.add_option(
             '--minion',
             '--minion-tests',