test: Automatically discover new signals.py files in tests

openedx_events/event_bus/avro/tests/test_avro.py

@@ -4,10 +4,6 @@
 
 from opaque_keys.edx.keys import CourseKey, UsageKey
 
-# Each new folder with signals must be manually imported in order for the signals to be cached
-#   and used in the unit tests. Using 'disable=reimported' with pylint will work,
-#   because we just use the cached signal list.
-from openedx_events.content_authoring import signals  # pylint: disable=unused-import
 from openedx_events.event_bus.avro.deserializer import AvroSignalDeserializer
 from openedx_events.event_bus.avro.serializer import AvroSignalSerializer
 from openedx_events.event_bus.avro.tests.test_utilities import (
@@ -20,8 +16,7 @@
     deserialize_bytes_to_event_data,
     serialize_event_data_to_bytes,
 )
-from openedx_events.learning import signals  # See note above; pylint: disable=reimported
-from openedx_events.tests.utils import FreezeSignalCacheMixin
+from openedx_events.tests.utils import FreezeSignalCacheMixin, load_all_signals
 from openedx_events.tooling import OpenEdxPublicSignal
 
 # If a signal is explicitly not for use with the event bus, add it to this list
@@ -71,6 +66,13 @@ def generate_test_event_data_for_data_type(data_type):
 
 class TestAvro(FreezeSignalCacheMixin, TestCase):
     """Tests for end-to-end serialization and deserialization of events"""
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        # Ensure we can usefully call all_events()
+        load_all_signals()
+
     def test_all_events(self):
         for signal in OpenEdxPublicSignal.all_events():
             if signal.event_type in KNOWN_UNSERIALIZABLE_SIGNALS:

openedx_events/tests/utils.py

@@ -1,6 +1,9 @@
 """
 Utils used by Open edX event tests.
 """
+import pkgutil
+from importlib import import_module
+
 from openedx_events.tooling import OpenEdxPublicSignal
 
 
@@ -124,3 +127,32 @@ def start_events_isolation(cls):
         cls().disable_all_events()
         cls().enable_events_by_type(*cls.ENABLED_OPENEDX_EVENTS)
         cls().allow_send_events_failure(*cls.ENABLED_OPENEDX_EVENTS)
+
+
+def load_all_signals():
+    """
+    Ensure OpenEdxPublicSignal.all_events() cache is fully populated.
+
+    Loads all non-test signals.py modules.
+    """
+    found = set()
+
+    root = import_module('openedx_events')
+    for m in pkgutil.walk_packages(root.__path__, root.__name__ + '.'):
+        module_name = m.name
+        if 'tests' in module_name.split('.') or '.test_' in module_name:
+            continue
+        if module_name.endswith('.signals'):
+            import_module(module_name)
+            found.add(module_name)
+
+    # Check that the auto-discovered list matches the known modules.
+    # This is just here to ensure that the auto-discovery is working
+    # properly and doesn't start to silently fail.
+    #
+    # If this assertion fails because a module has been added, renamed,
+    # or deleted, please update the hardcoded list.
+    assert found == {
+        'openedx_events.content_authoring.signals',
+        'openedx_events.learning.signals',
+    }