mirror of
https://github.com/apache/airflow.git
synced 2026-11-04 07:12:48 +00:00
Oleksii Maksymiv
49789f2846
__register_plugins used `return` instead of `continue` when encountering a duplicate plugin name, so the first duplicate abandoned the rest of the batch and every plugin after it was silently dropped. airflow-core carries a duplicate of this function and was fixed twice: in #60498 (`return` became a warning plus `continue`) and in #66649 (also record the duplicate in import_errors). Neither fix reached the Task SDK copy. Because that copy is the one used by the task-running process, plugins would load for the scheduler and API server but silently go missing inside tasks - most visibly, task instance listeners registered by a plugin would never fire. The __register_plugins body is now byte-identical to the airflow-core one.
113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
#
|
|
# Licensed to the Apache Software Foundation (ASF) under one
|
|
# or more contributor license agreements. See the NOTICE file
|
|
# distributed with this work for additional information
|
|
# regarding copyright ownership. The ASF licenses this file
|
|
# to you under the Apache License, Version 2.0 (the
|
|
# "License"); you may not use this file except in compliance
|
|
# with the License. You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing,
|
|
# software distributed under the License is distributed on an
|
|
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
# KIND, either express or implied. See the License for the
|
|
# specific language governing permissions and limitations
|
|
# under the License.
|
|
from __future__ import annotations
|
|
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from airflow.sdk import plugins_manager
|
|
from airflow.sdk.plugins_manager import AirflowPlugin
|
|
|
|
# ``_disable_ol_plugin`` in ``task-sdk/tests/conftest.py`` is a session-scoped autouse fixture that
|
|
# replaces ``_get_plugins`` with a stub returning no plugins. Capture the real, ``@cache``-wrapped
|
|
# function here at collection time — before any fixture runs — so these tests can put it back.
|
|
_REAL_GET_PLUGINS = plugins_manager._get_plugins
|
|
|
|
|
|
class TestGetPlugins:
|
|
"""Mirrors ``TestPluginsManager`` in ``airflow-core/tests/unit/plugins/test_plugins_manager.py``."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _use_real_get_plugins(self, monkeypatch):
|
|
monkeypatch.setattr(plugins_manager, "_get_plugins", _REAL_GET_PLUGINS)
|
|
_REAL_GET_PLUGINS.cache_clear()
|
|
yield
|
|
_REAL_GET_PLUGINS.cache_clear()
|
|
|
|
def test_duplicate_plugin_name_does_not_prevent_loading_subsequent_plugins(self):
|
|
"""
|
|
A duplicate name must skip that one plugin, not abandon the rest of the batch.
|
|
|
|
``__register_plugins`` used to ``return`` on the first duplicate, so every plugin
|
|
enumerated after it was dropped without an error. Here ``plugin_b`` is loaded from
|
|
the plugins directory and again from an entry point; ``plugin_c`` follows the
|
|
duplicate in that second batch and is the one that used to disappear.
|
|
"""
|
|
|
|
class PluginA(AirflowPlugin):
|
|
name = "plugin_a"
|
|
|
|
class PluginB(AirflowPlugin):
|
|
name = "plugin_b"
|
|
|
|
class PluginC(AirflowPlugin):
|
|
name = "plugin_c"
|
|
|
|
plugin_a = PluginA()
|
|
plugin_b = PluginB()
|
|
plugin_b_dup = PluginB()
|
|
plugin_c = PluginC()
|
|
|
|
with (
|
|
mock.patch.object(plugins_manager.settings, "PLUGINS_FOLDER", "/dev/null"),
|
|
mock.patch.object(plugins_manager.settings, "LAZY_LOAD_PROVIDERS", False),
|
|
mock.patch.object(
|
|
plugins_manager,
|
|
"_load_plugins_from_plugin_directory",
|
|
return_value=([plugin_a, plugin_b], {}),
|
|
),
|
|
mock.patch.object(
|
|
plugins_manager, "_load_entrypoint_plugins", return_value=([plugin_b_dup, plugin_c], {})
|
|
),
|
|
mock.patch.object(plugins_manager, "_load_providers_plugins", return_value=([], {})),
|
|
):
|
|
plugins, _ = plugins_manager._get_plugins()
|
|
|
|
plugin_names = [plugin.name for plugin in plugins]
|
|
assert "plugin_a" in plugin_names
|
|
assert "plugin_b" in plugin_names
|
|
assert "plugin_c" in plugin_names
|
|
assert len(plugins) == 3
|
|
|
|
def test_duplicate_plugin_name_is_reported_as_import_error(self):
|
|
"""The skipped duplicate is surfaced to the caller rather than dropped silently."""
|
|
|
|
class PluginA(AirflowPlugin):
|
|
name = "plugin_a"
|
|
|
|
class PluginADuplicateName(AirflowPlugin):
|
|
name = "plugin_a"
|
|
|
|
plugin_a = PluginA()
|
|
plugin_a_dup = PluginADuplicateName()
|
|
|
|
with (
|
|
mock.patch.object(plugins_manager.settings, "PLUGINS_FOLDER", "/dev/null"),
|
|
mock.patch.object(plugins_manager.settings, "LAZY_LOAD_PROVIDERS", False),
|
|
mock.patch.object(
|
|
plugins_manager, "_load_plugins_from_plugin_directory", return_value=([plugin_a], {})
|
|
),
|
|
mock.patch.object(plugins_manager, "_load_entrypoint_plugins", return_value=([plugin_a_dup], {})),
|
|
mock.patch.object(plugins_manager, "_load_providers_plugins", return_value=([], {})),
|
|
):
|
|
plugins, import_errors = plugins_manager._get_plugins()
|
|
|
|
assert [plugin.name for plugin in plugins] == ["plugin_a"]
|
|
assert len(import_errors) == 1
|