Fix module loading in logging config (#54555)

* fix(logging): fix module loading by using `import_module()`

- Replace `import_string()` with `import_module()` in `load_logging_config()`
- Fix issue where simple module paths like `log_config.LOGGING_CONFIG` failed
- Add `importlib` import inside function to avoid circular imports

* test(logging): add tests for logging config module path handling

- Add parametrized test for simple and nested module paths
- Add test for graceful fallback when remote logging vars are missing
- Create helper method to reduce test code duplication
- Test both log_config.LOGGING_CONFIG and nested.module.LOGGING_CONFIG scenarios

Ensures logging config works with various module path structures

* fix(logging): fix module loading by using `import_module()`

- Replace `import_string()` with `import_module()` in `load_logging_config()`
- Fix issue where simple module paths like `log_config.LOGGING_CONFIG` failed
- Add `importlib` import inside function to avoid circular imports

* test(logging): add tests for logging config module path handling

- Add parametrized test for simple and nested module paths
- Add test for graceful fallback when remote logging vars are missing
- Create helper method to reduce test code duplication
- Test both log_config.LOGGING_CONFIG and nested.module.LOGGING_CONFIG scenarios

Ensures logging config works with various module path structures

* fix(logging): fix module loading by using `import_module()`

- Replace `import_string()` with `import_module()` in `load_logging_config()`
- Fix issue where simple module paths like `log_config.LOGGING_CONFIG` failed
- Add `importlib` import inside function to avoid circular imports

* test(logging): add tests for logging config module path handling

- Add parametrized test for simple and nested module paths
- Add test for graceful fallback when remote logging vars are missing
- Create helper method to reduce test code duplication
- Test both log_config.LOGGING_CONFIG and nested.module.LOGGING_CONFIG scenarios

Ensures logging config works with various module path structures

* refactor(logging): move `import_module` to the top level

* test(logging): optimize test constants to reduce code duplication

Replace `SETTINGS_FILE_NESTED_MODULE` replicate part with string replacement from `SETTINGS_FILE_SIMPLE_MODULE`

* test(logging): use shared helper method for logging config validation

Replace manual assertions with `self._verify_basic_logging_config` in fallback test to ensure consistent validation across all logging config tests.

* refactor(logging_config): simplify, improve test config vars

- Remove redundant `SETTINGS_FILE_NESTED_MODULE`
- Rename `SETTINGS_FILE_SIMPLE_MODULE` to `SETTINGS_FILE_WITH_REMOTE_VARS`

* refactor(logging): imporve & simplify logging testing file

1. remove `SETTING_FILE_NO_REMOTE_VARS` bcz redundant var
2. add return typo with `_verify_basic_logging_config`
3. add unittest.mock `call`

Author: viiccwen

airflow-core/src/airflow/logging_config.py

@@ -19,6 +19,7 @@
 
 import logging
 import warnings
+from importlib import import_module
 from logging.config import dictConfig
 from typing import TYPE_CHECKING, Any
 
@@ -73,7 +74,9 @@ def load_logging_config() -> tuple[dict[str, Any], str]:
     else:
         modpath = logging_class_path.rsplit(".", 1)[0]
         try:
-            mod = import_string(modpath)
+            mod = import_module(modpath)
+
+            # Load remote logging configuration from the custom module
             REMOTE_TASK_LOG = getattr(mod, "REMOTE_TASK_LOG")
             DEFAULT_REMOTE_CONN_ID = getattr(mod, "DEFAULT_REMOTE_CONN_ID", None)
         except Exception as err:

airflow-core/tests/unit/core/test_logging_config.py

@@ -24,7 +24,7 @@
 import pathlib
 import sys
 import tempfile
-from unittest.mock import patch
+from unittest.mock import call, patch
 
 import pytest
 
@@ -96,6 +96,11 @@
 # Other settings here
 """
 
+SETTINGS_FILE_WITH_REMOTE_VARS = f"""{SETTINGS_FILE_VALID}
+REMOTE_TASK_LOG = None
+DEFAULT_REMOTE_CONN_ID = "test_conn_id"
+"""
+
 SETTINGS_DEFAULT_NAME = "custom_airflow_local_settings"
 
 
@@ -191,6 +196,15 @@ def teardown_method(self):
         importlib.reload(airflow_local_settings)
         configure_logging()
 
+    def _verify_basic_logging_config(
+        self, logging_config: dict, logging_class_path: str, expected_path: str
+    ) -> None:
+        """Helper method to verify basic logging config structure"""
+        assert isinstance(logging_config, dict)
+        assert logging_config["version"] == 1
+        assert "airflow.task" in logging_config["loggers"]
+        assert logging_class_path == expected_path
+
     # When we try to load an invalid config file, we expect an error
     def test_loading_invalid_local_settings(self):
         from airflow.logging_config import configure_logging, log
@@ -365,3 +379,44 @@ def test_loading_remote_logging_with_hdfs_handler(self):
             airflow.logging_config.configure_logging()
 
         assert isinstance(airflow.logging_config.REMOTE_TASK_LOG, HdfsRemoteLogIO)
+
+    @pytest.mark.parametrize(
+        "settings_content,module_structure,expected_path",
+        [
+            (SETTINGS_FILE_WITH_REMOTE_VARS, None, f"{SETTINGS_DEFAULT_NAME}.LOGGING_CONFIG"),
+            (
+                SETTINGS_FILE_WITH_REMOTE_VARS,
+                "nested.config.module",
+                f"nested.config.module.{SETTINGS_DEFAULT_NAME}.LOGGING_CONFIG",
+            ),
+        ],
+    )
+    def test_load_logging_config_module_paths(
+        self, settings_content: str, module_structure: str, expected_path: str
+    ):
+        """Test that load_logging_config works with different module path structures"""
+        dir_structure = module_structure.replace(".", "/") if module_structure else None
+
+        with settings_context(settings_content, dir_structure):
+            from airflow.logging_config import load_logging_config
+
+            logging_config, logging_class_path = load_logging_config()
+            self._verify_basic_logging_config(logging_config, logging_class_path, expected_path)
+
+    def test_load_logging_config_fallback_behavior(self):
+        """Test that load_logging_config falls back gracefully when remote logging vars are missing"""
+        with settings_context(SETTINGS_FILE_VALID):
+            from airflow.logging_config import load_logging_config
+
+            with patch("airflow.logging_config.log") as mock_log:
+                logging_config, logging_class_path = load_logging_config()
+
+                self._verify_basic_logging_config(
+                    logging_config, logging_class_path, f"{SETTINGS_DEFAULT_NAME}.LOGGING_CONFIG"
+                )
+
+                mock_log.info.mock_calls = [
+                    call(
+                        "Remote task logs will not be available due to an error:  %s",
+                    )
+                ]