Fix get_plugin_info for class based listeners.

airflow/plugins_manager.py

@@ -557,8 +557,10 @@ def get_plugin_info(attrs_to_dump: Iterable[str] | None = None) -> list[dict[str
                 elif attr in ("macros", "timetables", "hooks", "executors"):
                     info[attr] = [qualname(d) for d in getattr(plugin, attr)]
                 elif attr == "listeners":
-                    # listeners are always modules
-                    info[attr] = [d.__name__ for d in getattr(plugin, attr)]
+                    # listeners may be modules or class instances
+                    info[attr] = [
+                        d.__name__ if inspect.ismodule(d) else qualname(d) for d in getattr(plugin, attr)
+                    ]
                 elif attr == "appbuilder_views":
                     info[attr] = [
                         {**d, "view": qualname(d["view"].__class__) if "view" in d else None}

tests/cli/commands/test_plugins_command.py

@@ -85,7 +85,10 @@ def test_should_display_one_plugins(self):
                     "<tests.test_utils.mock_operators.CustomBaseIndexOpLink object>",
                 ],
                 "hooks": ["tests.plugins.test_plugin.PluginHook"],
-                "listeners": ["tests.listeners.empty_listener"],
+                "listeners": [
+                    "tests.listeners.empty_listener",
+                    "tests.listeners.class_listener.ClassBasedListener",
+                ],
                 "source": None,
                 "appbuilder_menu_items": [
                     {"name": "Google", "href": "https://www.google.com", "category": "Search"},

tests/plugins/test_plugin.py

@@ -32,6 +32,7 @@
 from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
 from airflow.timetables.interval import CronDataIntervalTimetable
 from tests.listeners import empty_listener
+from tests.listeners.class_listener import ClassBasedListener
 from tests.test_utils.mock_operators import (
     AirflowLink,
     AirflowLink2,
@@ -129,7 +130,7 @@ class AirflowTestPlugin(AirflowPlugin):
     ]
     operator_extra_links = [GoogleLink(), AirflowLink2(), CustomOpLink(), CustomBaseIndexOpLink(1)]
     timetables = [CustomCronDataIntervalTimetable]
-    listeners = [empty_listener]
+    listeners = [empty_listener, ClassBasedListener()]
     ti_deps = [CustomTestTriggerRule()]
 
 

tests/plugins/test_plugins_manager.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import importlib
+import inspect
 import logging
 import os
 import sys
@@ -29,6 +30,7 @@
 from airflow.hooks.base import BaseHook
 from airflow.listeners.listener import get_listener_manager
 from airflow.plugins_manager import AirflowPlugin
+from airflow.utils.module_loading import qualname
 from airflow.www import app as application
 from setup import AIRFLOW_SOURCES_ROOT
 from tests.test_utils.config import conf_vars
@@ -379,7 +381,13 @@ def test_registering_plugin_listeners(self):
             plugins_manager.integrate_listener_plugins(get_listener_manager())
 
             assert get_listener_manager().has_listeners
-            assert get_listener_manager().pm.get_plugins().pop().__name__ == "tests.listeners.empty_listener"
+            listeners = get_listener_manager().pm.get_plugins()
+            listener_names = [el.__name__ if inspect.ismodule(el) else qualname(el) for el in listeners]
+            # sort names as order of listeners is not guaranteed
+            assert [
+                "tests.listeners.class_listener.ClassBasedListener",
+                "tests.listeners.empty_listener",
+            ] == sorted(listener_names)
 
     def test_should_import_plugin_from_providers(self):
         from airflow import plugins_manager