Fix lookup_from_deprecated_options in AirflowConfigParser

airflow/cli/commands/remote_commands/config_command.py

@@ -522,7 +522,7 @@ def lint_config(args) -> None:
             continue
 
         if conf.has_option(
-            configuration.config.section, configuration.config.option, lookup_from_deprecated_options=False
+            configuration.config.section, configuration.config.option, lookup_from_deprecated=False
         ):
             lint_issues.append(configuration.message)
 

airflow/configuration.py

@@ -898,7 +898,7 @@ def get(  # type: ignore[override,misc]
         section: str,
         key: str,
         suppress_warnings: bool = False,
-        lookup_from_deprecated_options: bool = True,
+        lookup_from_deprecated: bool = True,
         _extra_stacklevel: int = 0,
         **kwargs,
     ) -> str | None:
@@ -908,8 +908,10 @@ def get(  # type: ignore[override,misc]
         deprecated_section: str | None = None
         deprecated_key: str | None = None
 
-        if lookup_from_deprecated_options:
-            option_description = self.configuration_description.get(section, {}).get(key, {})
+        if lookup_from_deprecated:
+            option_description = (
+                self.configuration_description.get(section, {}).get("options", {}).get(key, {})
+            )
             if option_description.get("deprecated"):
                 deprecation_reason = option_description.get("deprecation_reason", "")
                 warnings.warn(
@@ -1256,7 +1258,7 @@ def read_dict(  # type: ignore[override]
         """
         super().read_dict(dictionary=dictionary, source=source)
 
-    def has_option(self, section: str, option: str, lookup_from_deprecated_options: bool = True) -> bool:
+    def has_option(self, section: str, option: str, lookup_from_deprecated: bool = True) -> bool:
         """
         Check if option is defined.
 
@@ -1265,6 +1267,7 @@ def has_option(self, section: str, option: str, lookup_from_deprecated_options:
 
         :param section: section to get option from
         :param option: option to get
+        :param lookup_from_deprecated: If True, check if the option is defined in deprecated sections
         :return:
         """
         try:
@@ -1274,7 +1277,7 @@ def has_option(self, section: str, option: str, lookup_from_deprecated_options:
                 fallback=None,
                 _extra_stacklevel=1,
                 suppress_warnings=True,
-                lookup_from_deprecated_options=lookup_from_deprecated_options,
+                lookup_from_deprecated=lookup_from_deprecated,
             )
             if value is None:
                 return False

tests/cli/commands/remote_commands/test_config_command.py

@@ -324,7 +324,7 @@ def test_lint_with_invalid_section_option(self):
     def test_lint_detects_multiple_issues(self):
         with mock.patch(
             "airflow.configuration.conf.has_option",
-            side_effect=lambda section, option, lookup_from_deprecated_options: option
+            side_effect=lambda section, option, lookup_from_deprecated: option
             in ["check_slas", "strict_dataset_uri_validation"],
         ):
             with contextlib.redirect_stdout(StringIO()) as temp_stdout:

tests/core/test_configuration.py

@@ -938,6 +938,48 @@ def test_deprecated_options(self):
             with pytest.warns(DeprecationWarning), conf_vars({("celery", "celeryd_concurrency"): "99"}):
                 assert conf.getint("celery", "worker_concurrency") == 99
 
+    @pytest.mark.parametrize(
+        "deprecated_options_dict, kwargs, new_section_expected_value, old_section_expected_value",
+        [
+            pytest.param(
+                {("old_section", "old_key"): ("new_section", "new_key", "2.0.0")},
+                {"fallback": None},
+                None,
+                "value",
+                id="deprecated_in_different_section_lookup_enabled",
+            ),
+            pytest.param(
+                {("old_section", "old_key"): ("new_section", "new_key", "2.0.0")},
+                {"fallback": None, "lookup_from_deprecated": False},
+                None,
+                None,
+                id="deprecated_in_different_section_lookup_disabled",
+            ),
+            pytest.param(
+                {("new_section", "old_key"): ("new_section", "new_key", "2.0.0")},
+                {"fallback": None},
+                "value",
+                None,
+                id="deprecated_in_same_section_lookup_enabled",
+            ),
+            pytest.param(
+                {("new_section", "old_key"): ("new_section", "new_key", "2.0.0")},
+                {"fallback": None, "lookup_from_deprecated": False},
+                None,
+                None,
+                id="deprecated_in_same_section_lookup_disabled",
+            ),
+        ],
+    )
+    def test_deprecated_options_with_lookup_from_deprecated(
+        self, deprecated_options_dict, kwargs, new_section_expected_value, old_section_expected_value
+    ):
+        with conf_vars({("new_section", "new_key"): "value"}):
+            with set_deprecated_options(deprecated_options=deprecated_options_dict):
+                assert conf.get("new_section", "old_key", **kwargs) == new_section_expected_value
+
+                assert conf.get("old_section", "old_key", **kwargs) == old_section_expected_value
+
     @conf_vars(
         {
             ("celery", "result_backend"): None,