From 9151a4bc31bd800aa9e60cf13a55aaba9541b4e1 Mon Sep 17 00:00:00 2001
From: Jakub Dardzinski <kuba0221@gmail.com>
Date: Wed, 18 Oct 2023 15:48:58 +0200
Subject: [PATCH] Fix `get_plugin_info` for class based listeners.

Signed-off-by: Jakub Dardzinski <kuba0221@gmail.com>
---
 airflow/plugins_manager.py                 |  6 ++++--
 tests/cli/commands/test_plugins_command.py |  5 ++++-
 tests/plugins/test_plugin.py               |  3 ++-
 tests/plugins/test_plugins_manager.py      | 10 +++++++++-
 4 files changed, 19 insertions(+), 5 deletions(-)

diff --git a/airflow/plugins_manager.py b/airflow/plugins_manager.py
index beab0f3b071a2b..7275588d52e072 100644
--- a/airflow/plugins_manager.py
+++ b/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}
diff --git a/tests/cli/commands/test_plugins_command.py b/tests/cli/commands/test_plugins_command.py
index cff56fa3ead930..d180191bbc7cdc 100644
--- a/tests/cli/commands/test_plugins_command.py
+++ b/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"},
diff --git a/tests/plugins/test_plugin.py b/tests/plugins/test_plugin.py
index 6648215f3117ba..e207fd12da0043 100644
--- a/tests/plugins/test_plugin.py
+++ b/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()]
 
 
diff --git a/tests/plugins/test_plugins_manager.py b/tests/plugins/test_plugins_manager.py
index adc1507079f3d0..b0536e81a3f106 100644
--- a/tests/plugins/test_plugins_manager.py
+++ b/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