refactor(reports): address review feedback on Slack v1 deprecation

Code-review changes:

- Replace module-level `_v1_*_warning_emitted` booleans with `functools.cache`-
  decorated `_emit_v1_*_deprecation` helpers. Bare module globals had a
  read-then-write race under multi-threaded WSGI workers; functools.cache is
  thread-safe under the GIL and produces actually-once-per-process semantics
  without the noqa: PLW0603 escape hatch.
- Mention `groups:read` (in addition to `channels:read`) wherever the scope
  requirement appears: deprecation message constant, config.py comment, the
  scope-missing logger.warning, UPDATING.md, and (auto-synced) feature-flags.json.
  The v2 channel resolver queries both public_channel and private_channel types,
  so granting only `channels:read` silently breaks private-channel reports.
- Add `test_propagates_non_slack_api_errors_from_probe` — locks in that any
  exception other than SlackApiError (network, transport) propagates out of
  should_use_v2_api rather than masquerading as a missing-scope warning.
- Drop a tautological `assert_not_called()` on `get_channels_with_search` in
  the auto-upgrade round-trip test. SlackV2Notification.send() never calls that
  helper in any path, so the assertion was true by construction rather than
  by the test exercising a real fast path.
- Pin assertions on the deprecation-warning *message* to the exported
  `_SLACK_V1_DEPRECATION_MESSAGE` constant instead of substring fragments.
- Update the test autouse fixture to clear the new functools.cache caches
  rather than reset the now-removed module globals.

Three architectural concerns from review (auto-upgrade transaction race,
concurrent worker upgrade race, end-of-deprecation cleanup migration) are
pre-existing on the upgrade path and tracked as separate follow-up tasks
rather than expanded into this PR.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

UPDATING.md

@@ -24,6 +24,8 @@ assists people when migrating to a new version.
 
 ## Next
 
+- [39914](https://github.com/apache/superset/pull/39914) `ALERT_REPORT_SLACK_V2` now defaults to `True` and the legacy Slack v1 integration (`Slack` recipient type, `files.upload` API) is deprecated for removal in the next major. Slack retired `files.upload` in 2025, so v1 file-bearing sends already fail at the API level — only text-only `chat_postMessage` still works via the legacy path. Grant your Slack bot the `channels:read` scope (and `groups:read` if you use private channels) so existing `Slack` recipients can be auto-upgraded to `SlackV2` on next send. Operators who explicitly override the flag to `False` will see a one-shot `DeprecationWarning` plus a `logger.warning`; remove the override or grant the scopes to clear it.
+
 ### Granular Export Controls
 
 A new feature flag `GRANULAR_EXPORT_CONTROLS` introduces three fine-grained permissions that replace the legacy `can_csv` permission:

docs/static/feature-flags.json

@@ -116,9 +116,9 @@
       },
       {
         "name": "ALERT_REPORT_SLACK_V2",
-        "default": false,
+        "default": true,
         "lifecycle": "testing",
-        "description": "Enables Slack V2 integration for Alerts and Reports"
+        "description": "Enables Slack V2 integration for Alerts and Reports. Defaults to True; the legacy Slack v1 path is deprecated and will be removed in the next major release. Operators must grant the Slack bot the `channels:read` scope (and `groups:read` for private channels) so existing v1 recipients can be auto-upgraded on their next send. Without those scopes, file uploads fail (Slack retired the `files.upload` endpoint in 2025) and only text-only `chat_postMessage` sends will continue to work via the legacy path."
       },
       {
         "name": "ALERT_REPORT_WEBHOOK",

superset/config.py

@@ -611,9 +611,16 @@ DEFAULT_FEATURE_FLAGS: dict[str, bool] = {
     # @lifecycle: testing
     # @docs: https://superset.apache.org/docs/configuration/alerts-reports
     "ALERT_REPORTS": False,
-    # Enables Slack V2 integration for Alerts and Reports
+    # Enables Slack V2 integration for Alerts and Reports.
+    # Defaults to True; the legacy Slack v1 path is deprecated and will be removed
+    # in the next major release. Operators must grant the Slack bot the
+    # `channels:read` scope (and `groups:read` for private channels) so existing
+    # v1 recipients can be auto-upgraded on their next send. Without those
+    # scopes, file uploads fail (Slack retired the `files.upload` endpoint in
+    # 2025) and only text-only `chat_postMessage` sends will continue to work
+    # via the legacy path.
     # @lifecycle: testing
-    "ALERT_REPORT_SLACK_V2": False,
+    "ALERT_REPORT_SLACK_V2": True,
     # Enables webhook integration for Alerts and Reports
     # @lifecycle: testing
     "ALERT_REPORT_WEBHOOK": False,

superset/reports/notifications/slack.py

@@ -53,7 +53,11 @@ from superset.utils.slack import (
 logger = logging.getLogger(__name__)
 
 
-# TODO: Deprecated: Remove this class in Superset 6.0.0
+# Deprecated: Slack v1 will be removed in the next major release. The Slack
+# `files.upload` endpoint was retired in 2025, so file-bearing sends already
+# fail at the API level; only text-only `chat_postMessage` sends still work
+# here. Recipients with the `channels:read` scope are auto-upgraded to
+# SlackV2 on first send via `update_report_schedule_slack_v2`.
 class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-few-public-methods
     """
     Sends a slack notification for a report recipient

superset/reports/notifications/slackv2.py

@@ -77,7 +77,18 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
             return ("pdf", [self._content.pdf])
         return (None, [])
 
-    @backoff.on_exception(backoff.expo, SlackApiError, factor=10, base=2, max_tries=5)
+    # Retry on NotificationUnprocessableException (the wrapper that send()
+    # raises for transient Slack failures: SlackApiError, connection errors,
+    # and the SlackClientError catch-all). Retrying on SlackApiError directly
+    # would never fire because the try/except below converts it before the
+    # decorator can see it. Mirrors the pattern in webhook.py.
+    @backoff.on_exception(
+        backoff.expo,
+        NotificationUnprocessableException,
+        factor=10,
+        base=2,
+        max_tries=5,
+    )
     @statsd_gauge("reports.slack.send")
     def send(self) -> None:
         global_logs_context = getattr(g, "logs_context", {}) or {}

superset/utils/slack.py

@@ -16,7 +16,9 @@
 # under the License.
 
 
+import functools
 import logging
+import warnings
 from typing import Callable, Optional
 
 from flask import current_app as app
@@ -34,6 +36,33 @@ from superset.utils.core import recipients_string_to_list
 
 logger = logging.getLogger(__name__)
 
+_SLACK_V1_DEPRECATION_MESSAGE = (
+    "Slack v1 (the legacy `Slack` recipient type and `files.upload` API) is "
+    "deprecated and will be removed in the next major release. Slack retired "
+    "the `files.upload` endpoint in 2025, so v1 file uploads no longer work; "
+    "only text-only `chat_postMessage` sends still succeed. Grant your Slack "
+    "bot the `channels:read` (and `groups:read` if you use private channels) "
+    "scopes so existing v1 recipients can be auto-upgraded to SlackV2 on "
+    "their next send."
+)
+
+
+# functools.cache gives us a process-lifetime, thread-safe one-shot guard
+# without the read-then-write race that bare module globals would have under
+# multi-threaded WSGI workers. The cached return value (None) is irrelevant —
+# we only care that the body executes at most once per process.
+@functools.cache
+def _emit_v1_flag_off_deprecation() -> None:
+    warnings.warn(_SLACK_V1_DEPRECATION_MESSAGE, DeprecationWarning, stacklevel=3)
+    logger.warning(
+        "ALERT_REPORT_SLACK_V2 is disabled; %s", _SLACK_V1_DEPRECATION_MESSAGE
+    )
+
+
+@functools.cache
+def _emit_v1_scope_missing_deprecation() -> None:
+    warnings.warn(_SLACK_V1_DEPRECATION_MESSAGE, DeprecationWarning, stacklevel=3)
+
 
 class SlackChannelTypes(StrEnum):
     PUBLIC = "public_channel"
@@ -181,6 +210,7 @@ def get_channels_with_search(
 
 def should_use_v2_api() -> bool:
     if not feature_flag_manager.is_feature_enabled("ALERT_REPORT_SLACK_V2"):
+        _emit_v1_flag_off_deprecation()
         return False
     try:
         client = get_slack_client()
@@ -188,11 +218,14 @@ def should_use_v2_api() -> bool:
         logger.info("Slack API v2 is available")
         return True
     except SlackApiError:
-        # use the v1 api but warn with a deprecation message
+        # The DeprecationWarning fires once per process, but the actionable
+        # log line fires every send so operators see it in their report logs.
+        _emit_v1_scope_missing_deprecation()
         logger.warning(
-            """Your current Slack scopes are missing `channels:read`. Please add
-            this to your Slack app in order to continue using the v1 API. Support
-            for the old Slack API will be removed in Superset version 6.0.0."""
+            "Slack bot is missing the `channels:read` (and `groups:read` for "
+            "private channels) scope; falling back to the deprecated v1 API. "
+            "%s",
+            _SLACK_V1_DEPRECATION_MESSAGE,
         )
         return False
 

tests/integration_tests/reports/commands_tests.py

@@ -1975,11 +1975,13 @@ def test_slack_chart_alert_no_attachment(email_mock, create_alert_email_chart):
     "load_birth_names_dashboard_with_slices",
     "create_report_slack_chart",
 )
+@patch("superset.commands.report.execute.get_channels_with_search")
 @patch("superset.utils.slack.WebClient")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
 def test_slack_token_callable_chart_report(
     screenshot_mock,
     slack_client_mock_class,
+    get_channels_with_search_mock,
     create_report_slack_chart,
 ):
     """
@@ -1990,9 +1992,20 @@ def test_slack_token_callable_chart_report(
     channel_name = notification_targets[0]
     channel_id = "channel_id"
     slack_client_mock_class.return_value = Mock()
+    # should_use_v2_api() probes via conversations_list(); a non-erroring return
+    # is enough — it doesn't read the response body. The v2 upgrade then resolves
+    # channel names through get_channels_with_search, which we mock directly.
     slack_client_mock_class.return_value.conversations_list.return_value = {
         "channels": [{"id": channel_id, "name": channel_name}]
     }
+    get_channels_with_search_mock.return_value = [
+        {
+            "id": channel_id,
+            "name": channel_name,
+            "is_member": True,
+            "is_private": False,
+        }
+    ]
 
     slack_token_mock = Mock(return_value="cool_code")
     with patch.dict("flask.current_app.config", {"SLACK_API_TOKEN": slack_token_mock}):

tests/unit_tests/reports/notifications/slack_tests.py

@@ -16,16 +16,45 @@
 # under the License.
 
 import uuid
-from unittest.mock import MagicMock, patch
+from unittest.mock import call, MagicMock, patch
 
 import pandas as pd
 import pytest
-from slack_sdk.errors import SlackApiError
+from slack_sdk.errors import (
+    BotUserAccessError,
+    SlackApiError,
+    SlackClientConfigurationError,
+    SlackClientError,
+    SlackClientNotConnectedError,
+    SlackObjectFormationError,
+    SlackRequestError,
+    SlackTokenRotationError,
+)
 
+from superset.reports.notifications.exceptions import (
+    NotificationAuthorizationException,
+    NotificationMalformedException,
+    NotificationParamException,
+    NotificationUnprocessableException,
+)
 from superset.reports.notifications.slackv2 import SlackV2Notification
 from superset.utils.core import HeaderDataType
 
 
+@pytest.fixture(autouse=True)
+def _skip_backoff_sleep():
+    """Make any @backoff.on_exception retries instant.
+
+    SlackV2Notification.send() retries up to 5 times with `backoff.expo(factor=10,
+    base=2)` — that's ~150s of real sleep on a persistently-failing send. We
+    don't care about the wall-clock waits in unit tests; patching `time.sleep`
+    inside backoff's sync runner keeps the assertion semantics (call_count,
+    raised exception type) without the wait.
+    """
+    with patch("backoff._sync.time.sleep"):
+        yield
+
+
 @pytest.fixture
 def mock_header_data() -> HeaderDataType:
     return {
@@ -310,6 +339,10 @@ def test_send_slack(
     )
 
 
+@patch(
+    "superset.utils.slack.feature_flag_manager.is_feature_enabled",
+    return_value=False,
+)
 @patch("superset.reports.notifications.slack.g")
 @patch("superset.reports.notifications.slack.logger")
 @patch("superset.utils.slack.get_slack_client")
@@ -319,6 +352,7 @@ def test_send_slack_no_feature_flag(
     slack_client_mock_util: MagicMock,
     logger_mock: MagicMock,
     flask_global_mock: MagicMock,
+    feature_flag_mock: MagicMock,
     mock_header_data,
 ) -> None:
     # `superset.models.helpers`, a dependency of following imports,
@@ -330,7 +364,7 @@ def test_send_slack_no_feature_flag(
     execution_id = uuid.uuid4()
     flask_global_mock.logs_context = {"execution_id": execution_id}
     slack_client_mock.return_value.chat_postMessage.return_value = {"ok": True}
-    # scopes are valid but the feature flag is off. It should still run Slack v1
+    # Even with valid scopes, ALERT_REPORT_SLACK_V2=False forces the v1 path.
     slack_client_mock_util.return_value.conversations_list.return_value = {
         "channels": [{"id": "foo", "name": "bar"}]
     }
@@ -431,3 +465,488 @@ def test_slack_mixin_get_body_truncates_large_table(
     )
     body = notification._get_body(content=content)
     assert "(table was truncated)" in body
+
+
+# ---------------------------------------------------------------------------
+# Bulletproof v2 send-path coverage
+#
+# The tests above exercise the chat_postMessage path (text-only sends). The
+# tests below cover files_upload_v2 across screenshots/CSV/PDF, multi-channel
+# fan-out, exception mapping, backoff, statsd, and logs propagation. Together
+# they guarantee that every observable behavior of SlackV2Notification.send()
+# is locked down before Slack v1 is removed.
+# ---------------------------------------------------------------------------
+
+
+def _make_v2_notification(content, target: str = "C12345"):
+    """Helper to build a SlackV2Notification with a given target string."""
+    from superset.reports.models import ReportRecipients, ReportRecipientType
+
+    return SlackV2Notification(
+        recipient=ReportRecipients(
+            type=ReportRecipientType.SLACKV2,
+            recipient_config_json=f'{{"target": "{target}"}}',
+        ),
+        content=content,
+    )
+
+
+def _make_content(mock_header_data, **overrides):
+    """Helper to build a minimal NotificationContent."""
+    from superset.reports.notifications.base import NotificationContent
+
+    defaults: dict = {
+        "name": "test alert",
+        "header_data": mock_header_data,
+        "description": "desc",
+    }
+    defaults.update(overrides)
+    return NotificationContent(**defaults)
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_with_single_screenshot_calls_files_upload_v2(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    flask_global_mock.logs_context = {"execution_id": uuid.uuid4()}
+    content = _make_content(mock_header_data, screenshots=[b"screenshot-bytes"])
+    notification = _make_v2_notification(content, target="C12345")
+
+    notification.send()
+
+    upload = slack_client_mock.return_value.files_upload_v2
+    upload.assert_called_once()
+    kwargs = upload.call_args.kwargs
+    assert kwargs["channel"] == "C12345"
+    assert kwargs["file"] == b"screenshot-bytes"
+    assert kwargs["title"] == "test alert"
+    assert kwargs["filename"] == "test alert.png"
+    assert "test alert" in kwargs["initial_comment"]
+    # chat_postMessage should NOT be called when files are present
+    slack_client_mock.return_value.chat_postMessage.assert_not_called()
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_with_multiple_screenshots_uploads_each(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    flask_global_mock.logs_context = {}
+    content = _make_content(
+        mock_header_data, screenshots=[b"shot-1", b"shot-2", b"shot-3"]
+    )
+    notification = _make_v2_notification(content, target="C12345")
+
+    notification.send()
+
+    upload = slack_client_mock.return_value.files_upload_v2
+    assert upload.call_count == 3
+    uploaded_files = [c.kwargs["file"] for c in upload.call_args_list]
+    assert uploaded_files == [b"shot-1", b"shot-2", b"shot-3"]
+    # All three uploads target the same single channel
+    for c in upload.call_args_list:
+        assert c.kwargs["channel"] == "C12345"
+        assert c.kwargs["filename"] == "test alert.png"
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_with_csv_calls_files_upload_v2(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    flask_global_mock.logs_context = {}
+    content = _make_content(mock_header_data, csv=b"col1,col2\n1,2\n")
+    notification = _make_v2_notification(content, target="C12345")
+
+    notification.send()
+
+    upload = slack_client_mock.return_value.files_upload_v2
+    upload.assert_called_once()
+    kwargs = upload.call_args.kwargs
+    assert kwargs["file"] == b"col1,col2\n1,2\n"
+    assert kwargs["filename"] == "test alert.csv"
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_with_pdf_calls_files_upload_v2(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    flask_global_mock.logs_context = {}
+    content = _make_content(mock_header_data, pdf=b"%PDF-1.4...")
+    notification = _make_v2_notification(content, target="C12345")
+
+    notification.send()
+
+    upload = slack_client_mock.return_value.files_upload_v2
+    upload.assert_called_once()
+    kwargs = upload.call_args.kwargs
+    assert kwargs["file"] == b"%PDF-1.4..."
+    assert kwargs["filename"] == "test alert.pdf"
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_to_multiple_channels_uploads_per_channel(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    flask_global_mock.logs_context = {}
+    content = _make_content(mock_header_data, screenshots=[b"shot-1", b"shot-2"])
+    notification = _make_v2_notification(content, target="C12345,C67890,C11111")
+
+    notification.send()
+
+    upload = slack_client_mock.return_value.files_upload_v2
+    # 3 channels x 2 files = 6 uploads
+    assert upload.call_count == 6
+    seen = {(c.kwargs["channel"], c.kwargs["file"]) for c in upload.call_args_list}
+    assert seen == {
+        ("C12345", b"shot-1"),
+        ("C12345", b"shot-2"),
+        ("C67890", b"shot-1"),
+        ("C67890", b"shot-2"),
+        ("C11111", b"shot-1"),
+        ("C11111", b"shot-2"),
+    }
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_text_only_uses_chat_post_message(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    flask_global_mock.logs_context = {}
+    content = _make_content(mock_header_data)
+    notification = _make_v2_notification(content, target="C12345,C67890")
+
+    notification.send()
+
+    # No files → chat_postMessage per channel, no files_upload_v2 calls
+    slack_client_mock.return_value.files_upload_v2.assert_not_called()
+    chat = slack_client_mock.return_value.chat_postMessage
+    assert chat.call_count == 2
+    channels = sorted(c.kwargs["channel"] for c in chat.call_args_list)
+    assert channels == ["C12345", "C67890"]
+
+
+def test_v2_inline_files_precedence(mock_header_data) -> None:
+    """CSV beats screenshots beats PDF; only one inline-file type is sent."""
+    content = _make_content(
+        mock_header_data,
+        csv=b"a,b\n1,2",
+        screenshots=[b"shot-1"],
+        pdf=b"%PDF",
+    )
+    notification = _make_v2_notification(content, target="C12345")
+    file_type, files = notification._get_inline_files()
+    assert file_type == "csv"
+    assert files == [b"a,b\n1,2"]
+
+    content = _make_content(
+        mock_header_data,
+        screenshots=[b"shot-1"],
+        pdf=b"%PDF",
+    )
+    notification = _make_v2_notification(content, target="C12345")
+    file_type, files = notification._get_inline_files()
+    assert file_type == "png"
+    assert files == [b"shot-1"]
+
+    content = _make_content(mock_header_data, pdf=b"%PDF")
+    notification = _make_v2_notification(content, target="C12345")
+    file_type, files = notification._get_inline_files()
+    assert file_type == "pdf"
+    assert files == [b"%PDF"]
+
+
+@pytest.mark.parametrize(
+    ("slack_exc_factory", "expected_exc"),
+    [
+        (
+            lambda: BotUserAccessError("bot user blocked"),
+            NotificationParamException,
+        ),
+        (
+            lambda: SlackRequestError("bad request"),
+            NotificationParamException,
+        ),
+        (
+            lambda: SlackClientConfigurationError("misconfigured"),
+            NotificationParamException,
+        ),
+        (
+            lambda: SlackObjectFormationError("malformed"),
+            NotificationMalformedException,
+        ),
+        (
+            lambda: SlackTokenRotationError(
+                SlackApiError(message="rotation failed", response={"ok": False})
+            ),
+            NotificationAuthorizationException,
+        ),
+        (
+            lambda: SlackClientNotConnectedError("offline"),
+            NotificationUnprocessableException,
+        ),
+        (
+            # Fallback: any other SlackClientError becomes Unprocessable.
+            lambda: SlackClientError("misc client error"),
+            NotificationUnprocessableException,
+        ),
+    ],
+)
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_maps_slack_sdk_exceptions(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    slack_exc_factory,
+    expected_exc,
+    mock_header_data,
+) -> None:
+    flask_global_mock.logs_context = {}
+    slack_client_mock.return_value.chat_postMessage.side_effect = slack_exc_factory()
+
+    content = _make_content(mock_header_data)
+    notification = _make_v2_notification(content, target="C12345")
+
+    with pytest.raises(expected_exc):
+        notification.send()
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_retries_on_transient_slack_api_error(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    """`@backoff.on_exception(NotificationUnprocessableException, max_tries=5)`
+    retries the wrapped exception that send() actually raises.
+
+    A persistent Slack rate-limit (or any other transient failure that maps to
+    NotificationUnprocessableException) results in exactly max_tries=5 send
+    attempts before the final exception propagates. This mirrors the existing
+    pattern in webhook.py.
+    """
+    flask_global_mock.logs_context = {}
+    slack_client_mock.return_value.chat_postMessage.side_effect = SlackApiError(
+        message="rate limited", response={"ok": False, "error": "ratelimited"}
+    )
+
+    content = _make_content(mock_header_data)
+    notification = _make_v2_notification(content, target="C12345")
+
+    with pytest.raises(NotificationUnprocessableException):
+        notification.send()
+
+    assert slack_client_mock.return_value.chat_postMessage.call_count == 5
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_does_not_retry_param_errors(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    """Non-transient errors (config / auth / malformed) are NOT retried — only
+    NotificationUnprocessableException triggers backoff. A
+    NotificationParamException-class failure (BotUserAccessError → 422) hits
+    the API exactly once and surfaces immediately.
+    """
+    flask_global_mock.logs_context = {}
+    slack_client_mock.return_value.chat_postMessage.side_effect = BotUserAccessError(
+        "bot user blocked"
+    )
+
+    content = _make_content(mock_header_data)
+    notification = _make_v2_notification(content, target="C12345")
+
+    with pytest.raises(NotificationParamException):
+        notification.send()
+
+    assert slack_client_mock.return_value.chat_postMessage.call_count == 1
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_records_statsd_gauge_on_success(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    flask_global_mock.logs_context = {}
+    content = _make_content(mock_header_data)
+    notification = _make_v2_notification(content, target="C12345")
+
+    with patch(
+        "superset.extensions.stats_logger_manager.instance.gauge"
+    ) as statsd_mock:
+        notification.send()
+
+    statsd_mock.assert_called_with("reports.slack.send.ok", 1)
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_records_statsd_gauge_warning_on_param_error(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    """Status<500 exceptions (NotificationParamException is 422) → .warning."""
+    flask_global_mock.logs_context = {}
+    slack_client_mock.return_value.chat_postMessage.side_effect = (
+        SlackClientConfigurationError("bad config")
+    )
+
+    content = _make_content(mock_header_data)
+    notification = _make_v2_notification(content, target="C12345")
+
+    with patch(
+        "superset.extensions.stats_logger_manager.instance.gauge"
+    ) as statsd_mock:
+        with pytest.raises(NotificationParamException):
+            notification.send()
+
+    assert call("reports.slack.send.warning", 1) in statsd_mock.call_args_list
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.logger")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_propagates_execution_id_to_logs(
+    slack_client_mock: MagicMock,
+    logger_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    """The success log carries the execution_id from g.logs_context."""
+    execution_id = uuid.uuid4()
+    flask_global_mock.logs_context = {"execution_id": execution_id}
+
+    content = _make_content(mock_header_data, screenshots=[b"shot"])
+    notification = _make_v2_notification(content, target="C12345")
+    notification.send()
+
+    logger_mock.info.assert_called_with(
+        "Report sent to slack", extra={"execution_id": execution_id}
+    )
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.logger")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_handles_missing_logs_context(
+    slack_client_mock: MagicMock,
+    logger_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    """When g.logs_context is None or missing, the log uses execution_id=None."""
+    # Mirrors `getattr(g, "logs_context", {}) or {}` falsy-coalescing path.
+    flask_global_mock.logs_context = None
+
+    content = _make_content(mock_header_data)
+    notification = _make_v2_notification(content, target="C12345")
+    notification.send()
+
+    logger_mock.info.assert_called_with(
+        "Report sent to slack", extra={"execution_id": None}
+    )
+
+
+# ---------------------------------------------------------------------------
+# End-to-end auto-upgrade: v1 recipient → SlackV1NotificationError → upgrade →
+# row mutated to IDs → second send takes the v2 fast path with no resolution.
+# ---------------------------------------------------------------------------
+
+
+@patch(
+    "superset.utils.slack.feature_flag_manager.is_feature_enabled",
+    return_value=True,
+)
+@patch("superset.reports.notifications.slack.g")
+@patch("superset.utils.slack.get_slack_client")
+@patch("superset.reports.notifications.slack.get_slack_client")
+@patch("superset.commands.report.execute.get_channels_with_search")
+def test_auto_upgrade_round_trip_v1_to_v2(
+    get_channels_with_search_mock: MagicMock,
+    v1_client_mock: MagicMock,
+    util_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    feature_flag_mock: MagicMock,
+    mock_header_data,
+) -> None:
+    """A v1 SLACK recipient with channel names is auto-upgraded to SlackV2 with
+    channel IDs after the first send raises SlackV1NotificationError.
+
+    The second send-as-v2 then uses the resolved IDs without further lookups.
+    """
+    from superset.commands.report.execute import BaseReportState
+    from superset.reports.models import (
+        ReportRecipients,
+        ReportRecipientType,
+        ReportSchedule,
+    )
+    from superset.reports.notifications.exceptions import SlackV1NotificationError
+    from superset.reports.notifications.slack import SlackNotification
+
+    flask_global_mock.logs_context = {}
+    # Scopes are present → should_use_v2_api returns True → v1.send raises
+    util_client_mock.return_value.conversations_list.return_value = {
+        "channels": [{"id": "C12345", "name": "general"}]
+    }
+    get_channels_with_search_mock.return_value = [
+        {"id": "C12345", "name": "general", "is_member": True, "is_private": False}
+    ]
+
+    schedule = ReportSchedule(
+        recipients=[
+            ReportRecipients(
+                type=ReportRecipientType.SLACK,
+                recipient_config_json='{"target": "general"}',
+            )
+        ]
+    )
+    content = _make_content(mock_header_data)
+
+    # Step 1: v1 send raises SlackV1NotificationError because v2 is available.
+    v1 = SlackNotification(recipient=schedule.recipients[0], content=content)
+    with pytest.raises(SlackV1NotificationError):
+        v1.send()
+
+    # Step 2: command-level upgrade rewrites the row to SlackV2 with channel IDs.
+    state = BaseReportState(schedule, "January 1, 2021", "exec-id")
+    state.update_report_schedule_slack_v2()
+    assert schedule.recipients[0].type == ReportRecipientType.SLACKV2
+    assert schedule.recipients[0].recipient_config_json == '{"target": "C12345"}'
+
+    # Step 3: a fresh SlackV2Notification on the rewritten row sends to the
+    # resolved channel ID directly. (SlackV2Notification.send() never calls
+    # get_channels_with_search itself, so the upgraded row carrying IDs in
+    # its target is the load-bearing assertion here.)
+    with patch(
+        "superset.reports.notifications.slackv2.get_slack_client"
+    ) as v2_client_mock:
+        v2 = SlackV2Notification(recipient=schedule.recipients[0], content=content)
+        v2.send()
+        v2_client_mock.return_value.chat_postMessage.assert_called_once()
+        assert (
+            v2_client_mock.return_value.chat_postMessage.call_args.kwargs["channel"]
+            == "C12345"
+        )

tests/unit_tests/utils/slack_test.py

@@ -15,9 +15,19 @@
 # specific language governing permissions and limitations
 # under the License.
 
-import pytest
+import warnings
 
-from superset.utils.slack import get_channels_with_search, SlackChannelTypes
+import pytest
+from slack_sdk.errors import SlackApiError, SlackClientNotConnectedError
+
+from superset.utils.slack import (
+    _emit_v1_flag_off_deprecation,
+    _emit_v1_scope_missing_deprecation,
+    _SLACK_V1_DEPRECATION_MESSAGE,
+    get_channels_with_search,
+    should_use_v2_api,
+    SlackChannelTypes,
+)
 
 
 class MockResponse:
@@ -216,3 +226,117 @@ The server responded with: missing scope: channels:read"""
             {"name": "general", "id": "C12345"},
             {"name": "random", "id": "C67890"},
         ]
+
+
+# ---------------------------------------------------------------------------
+# should_use_v2_api: drives the v1→v2 auto-upgrade decision and emits
+# DeprecationWarning + logger.warning for both no-flag and missing-scope cases.
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def _reset_v1_warning_caches():
+    """Each test sees fresh once-per-process warning state.
+
+    The deprecation emitters are wrapped in `functools.cache` to give
+    thread-safe one-shot semantics in production. Tests need them to fire
+    again, so we clear the cache before and after each case.
+    """
+    _emit_v1_flag_off_deprecation.cache_clear()
+    _emit_v1_scope_missing_deprecation.cache_clear()
+    yield
+    _emit_v1_flag_off_deprecation.cache_clear()
+    _emit_v1_scope_missing_deprecation.cache_clear()
+
+
+class TestShouldUseV2Api:
+    def test_returns_true_when_flag_on_and_scopes_present(self, mocker):
+        mocker.patch(
+            "superset.utils.slack.feature_flag_manager.is_feature_enabled",
+            return_value=True,
+        )
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = {
+            "channels": [{"id": "C1", "name": "general"}]
+        }
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            assert should_use_v2_api() is True
+            assert not any(issubclass(w.category, DeprecationWarning) for w in caught)
+
+    def test_returns_false_when_flag_off_and_emits_deprecation_once(self, mocker):
+        mocker.patch(
+            "superset.utils.slack.feature_flag_manager.is_feature_enabled",
+            return_value=False,
+        )
+        logger_mock = mocker.patch("superset.utils.slack.logger")
+
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            assert should_use_v2_api() is False
+            assert should_use_v2_api() is False  # second call: no new warning
+            assert should_use_v2_api() is False  # third call: no new warning
+
+        deprecation_warnings = [
+            w for w in caught if issubclass(w.category, DeprecationWarning)
+        ]
+        # Exactly one DeprecationWarning across three calls.
+        assert len(deprecation_warnings) == 1
+        assert str(deprecation_warnings[0].message) == _SLACK_V1_DEPRECATION_MESSAGE
+        # logger.warning fires only once for the same reason.
+        assert logger_mock.warning.call_count == 1
+        assert (
+            "ALERT_REPORT_SLACK_V2 is disabled" in logger_mock.warning.call_args.args[0]
+        )
+
+    def test_returns_false_when_scope_missing_and_emits_deprecation_once(self, mocker):
+        mocker.patch(
+            "superset.utils.slack.feature_flag_manager.is_feature_enabled",
+            return_value=True,
+        )
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = SlackApiError(
+            message="missing_scope", response={"ok": False}
+        )
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+        logger_mock = mocker.patch("superset.utils.slack.logger")
+
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            assert should_use_v2_api() is False
+            assert should_use_v2_api() is False
+            assert should_use_v2_api() is False
+
+        deprecation_warnings = [
+            w for w in caught if issubclass(w.category, DeprecationWarning)
+        ]
+        # DeprecationWarning emitted exactly once across multiple calls.
+        assert len(deprecation_warnings) == 1
+        assert str(deprecation_warnings[0].message) == _SLACK_V1_DEPRECATION_MESSAGE
+        # The user-visible scope-missing log fires every time, since operators
+        # need to see the actionable message in their report-execution logs.
+        assert logger_mock.warning.call_count == 3
+        for c in logger_mock.warning.call_args_list:
+            assert "channels:read" in c.args[0]
+            assert "groups:read" in c.args[0]
+
+    def test_propagates_non_slack_api_errors_from_probe(self, mocker):
+        """Any non-`SlackApiError` exception from the probe (network issue,
+        unexpected SDK error) propagates out of `should_use_v2_api` rather than
+        silently falling back to v1. Falling back on a non-API error would
+        mask real bugs as "you don't have channels:read", which is misleading.
+        """
+        mocker.patch(
+            "superset.utils.slack.feature_flag_manager.is_feature_enabled",
+            return_value=True,
+        )
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = SlackClientNotConnectedError(
+            "transport closed"
+        )
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        with pytest.raises(SlackClientNotConnectedError):
+            should_use_v2_api()