test(tasks): Add tests for log_task_failure signal handler

Added unit tests to verify the log_task_failure signal handler:
- Test logging with sender task provided
- Test logging when sender is None (fallback to "Unknown")

Both tests verify the correct logging behavior with proper exception info.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

superset/tasks/scheduler.py

@@ -22,6 +22,7 @@ from typing import Any
 
 from celery import Task
 from celery.exceptions import SoftTimeLimitExceeded
+from celery.signals import task_failure
 from flask import current_app
 
 from superset import is_feature_enabled
@@ -41,8 +42,32 @@ from superset.utils.log import get_logger_from_status
 logger = logging.getLogger(__name__)
 
 
-@celery_app.task(name="reports.scheduler")
-def scheduler() -> None:
+@task_failure.connect
+def log_task_failure(  # pylint: disable=unused-argument
+    sender: Task | None = None,
+    task_id: str | None = None,
+    exception: Exception | None = None,
+    args: tuple[Any, ...] | None = None,
+    kwargs: dict[str, Any] | None = None,
+    traceback: Any = None,
+    einfo: Any = None,
+    **kw: Any,
+) -> None:
+    task_name = sender.name if sender else "Unknown"
+    logger.exception("Celery task %s failed: %s", task_name, exception, exc_info=einfo)
+
+
+@celery_app.task(
+    name="reports.scheduler",
+    bind=True,
+    autoretry_for=(Exception,),
+    retry_kwargs={
+        "max_retries": 3,
+        "countdown": 60,
+    },  # Retry up to 3 times, wait 60s between
+    retry_backoff=True,  # exponential backoff
+)
+def scheduler(self: Task) -> None:  # pylint: disable=unused-argument
     """
     Celery beat main scheduler for reports
     """

tests/integration_tests/reports/scheduler_tests.py

@@ -16,7 +16,7 @@
 # under the License.
 
 from random import randint
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 import pytest
 from flask_appbuilder.security.sqla.models import User
@@ -25,7 +25,7 @@ from freezegun.api import FakeDatetime
 
 from superset.extensions import db
 from superset.reports.models import ReportScheduleType
-from superset.tasks.scheduler import execute, scheduler
+from superset.tasks.scheduler import execute, log_task_failure, scheduler
 from tests.integration_tests.reports.utils import insert_report_schedule
 from tests.integration_tests.test_app import app
 
@@ -201,3 +201,48 @@ def test_execute_task_with_command_exception(
 
     db.session.delete(report_schedule)
     db.session.commit()
+
+
+@patch("superset.tasks.scheduler.logger")
+def test_log_task_failure_with_sender(logger_mock):
+    """
+    Test that log_task_failure logs correctly when sender is provided
+    """
+    mock_task = MagicMock()
+    mock_task.name = "test.task.name"
+    mock_exception = Exception("Test error")
+    mock_einfo = MagicMock()
+
+    log_task_failure(
+        sender=mock_task,
+        task_id="test-task-id",
+        exception=mock_exception,
+        einfo=mock_einfo,
+    )
+
+    logger_mock.exception.assert_called_once_with(
+        "Celery task %s failed: %s",
+        "test.task.name",
+        mock_exception,
+        exc_info=mock_einfo,
+    )
+
+
+@patch("superset.tasks.scheduler.logger")
+def test_log_task_failure_without_sender(logger_mock):
+    """
+    Test that log_task_failure logs correctly when sender is None
+    """
+    mock_exception = Exception("Test error")
+    mock_einfo = MagicMock()
+
+    log_task_failure(
+        sender=None,
+        task_id="test-task-id",
+        exception=mock_exception,
+        einfo=mock_einfo,
+    )
+
+    logger_mock.exception.assert_called_once_with(
+        "Celery task %s failed: %s", "Unknown", mock_exception, exc_info=mock_einfo
+    )