kaxil commented on code in PR #65412:
URL: https://github.com/apache/airflow/pull/65412#discussion_r3100393541
##########
airflow-core/src/airflow/config_templates/airflow_local_settings.py:
##########
@@ -370,4 +370,4 @@ def _default_conn_name_from(mod_path, hook_name):
"section 'elasticsearch' if you are using Elasticsearch. In the
other case, "
"'remote_base_log_folder' option in the 'logging' section."
)
-
DEFAULT_LOGGING_CONFIG["handlers"]["task"].update(remote_task_handler_kwargs)
+ DEFAULT_LOGGING_CONFIG["handlers"]["task"].update(_handler_kwargs)
Review Comment:
import importlib
import json
from unittest import mock
import pytest
from airflow.config_templates import airflow_local_settings
from tests_common.test_utils.config import conf_vars
@pytest.fixture
def restore_local_settings():
yield
importlib.reload(airflow_local_settings)
@pytest.mark.parametrize(
"remote_base, remote_io_path",
[
("s3://bucket/path",
"airflow.providers.amazon.aws.log.s3_task_handler.S3RemoteLogIO"),
("wasb-logs",
"airflow.providers.microsoft.azure.log.wasb_task_handler.WasbRemoteLogIO"),
("gs://bucket/path",
"airflow.providers.google.cloud.log.gcs_task_handler.GCSRemoteLogIO"),
("cloudwatch://arn:aws:logs:us-east-1:0:log-group:foo",
"airflow.providers.amazon.aws.log.cloudwatch_task_handler.CloudWatchRemoteLogIO"),
("oss://bucket/path",
"airflow.providers.alibaba.cloud.log.oss_task_handler.OSSRemoteLogIO"),
("hdfs://host/path",
"airflow.providers.apache.hdfs.log.hdfs_task_handler.HdfsRemoteLogIO"),
],
ids=["s3", "wasb", "gcs", "cloudwatch", "oss", "hdfs"],
)
def test_remote_task_handler_kwargs_not_leaked_to_local_task_handler(
remote_base, remote_io_path, restore_local_settings
):
"""Regression for apache/airflow#65412: remote_task_handler_kwargs must be
consumed
by the RemoteLogIO for the backend, not re-applied to the local
FileTaskHandler."""
user_kwargs = {"remote_base": "ignored", "custom_key": "v"}
with mock.patch(remote_io_path), conf_vars(
{
("logging", "remote_logging"): "True",
("logging", "remote_base_log_folder"): remote_base,
("logging", "remote_task_handler_kwargs"): json.dumps(user_kwargs),
}
):
importlib.reload(airflow_local_settings)
task_cfg =
airflow_local_settings.DEFAULT_LOGGING_CONFIG["handlers"]["task"]
for k in user_kwargs:
assert k not in task_cfg, f"{k!r} leaked into task handler for
{remote_base}"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
