feat: add slackv2 notification (#29264)

2024-07-17 15:14:32 -07:00 · 2024-07-17 15:14:32 -07:00 · 6dbfe2aab9
parent c0d46eb1af
commit 6dbfe2aab9
33 changed files with 1667 additions and 556 deletions
--- a/UPDATING.md
+++ b/UPDATING.md
@ -57,6 +57,7 @@ assists people when migrating to a new version.
  translations inside the python package. This includes the .mo files needed by pybabel on the
  backend, as well as the .json files used by the frontend. If you were doing anything before
  as part of your bundling to expose translation packages, it's probably not needed anymore.
+- [29264](https://github.com/apache/superset/pull/29264) Slack has updated its file upload api, and we are now supporting this new api in Superset, although the Slack api is not backward compatible. The original Slack integration is deprecated and we will require a new Slack scope `channels:read` to be added to Slack workspaces in order to use this new api. In an upcoming release, we will make this new Slack scope mandatory and remove the old Slack functionality.

 ### Potential Downtime

--- a/superset-frontend/packages/superset-ui-core/src/utils/featureFlags.ts
+++ b/superset-frontend/packages/superset-ui-core/src/utils/featureFlags.ts
@ -25,6 +25,7 @@ export enum FeatureFlag {
  AlertsAttachReports = 'ALERTS_ATTACH_REPORTS',
  AlertReports = 'ALERT_REPORTS',
  AlertReportTabs = 'ALERT_REPORT_TABS',
+  AlertReportSlackV2 = 'ALERT_REPORT_SLACK_V2',
  AllowFullCsvExport = 'ALLOW_FULL_CSV_EXPORT',
  AvoidColorsCollision = 'AVOID_COLORS_COLLISION',
  ChartPluginsExperimental = 'CHART_PLUGINS_EXPERIMENTAL',
--- a/superset-frontend/src/components/Select/Select.test.tsx
+++ b/superset-frontend/src/components/Select/Select.test.tsx
@ -188,6 +188,20 @@ test('does not add a new option if the value is already in the options', async (
  expect(options).toHaveLength(1);
 });

+test('does not add new options when the value is in a nested/grouped option', async () => {
+  const options = [
+    {
+      label: 'Group',
+      options: [OPTIONS[0]],
+    },
+  ];
+  render(<Select {...defaultProps} options={options} value={OPTIONS[0]} />);
+  await open();
+  expect(await findSelectOption(OPTIONS[0].label)).toBeInTheDocument();
+  const selectOptions = await findAllSelectOptions();
+  expect(selectOptions).toHaveLength(1);
+});
+
 test('inverts the selection', async () => {
  render(<Select {...defaultProps} invertSelection />);
  await open();
--- a/superset-frontend/src/components/Select/Select.tsx
+++ b/superset-frontend/src/components/Select/Select.tsx
@ -182,8 +182,18 @@ const Select = forwardRef(

    // add selected values to options list if they are not in it
    const fullSelectOptions = useMemo(() => {
+      // check to see if selectOptions are grouped
+      let groupedOptions: SelectOptionsType;
+      if (selectOptions.some(opt => opt.options)) {
+        groupedOptions = selectOptions.reduce(
+          (acc, group) => [...acc, ...group.options],
+          [] as SelectOptionsType,
+        );
+      }
      const missingValues: SelectOptionsType = ensureIsArray(selectValue)
-        .filter(opt => !hasOption(getValue(opt), selectOptions))
+        .filter(
+          opt => !hasOption(getValue(opt), groupedOptions || selectOptions),
+        )
        .map(opt =>
          isLabeledValue(opt) ? opt : { value: opt, label: String(opt) },
        );
--- a/superset-frontend/src/features/alerts/AlertReportModal.test.tsx
+++ b/superset-frontend/src/features/alerts/AlertReportModal.test.tsx
@ -21,7 +21,7 @@ import fetchMock from 'fetch-mock';
 import { render, screen, waitFor, within } from 'spec/helpers/testing-library';
 import { buildErrorTooltipMessage } from './buildErrorTooltipMessage';
 import AlertReportModal, { AlertReportModalProps } from './AlertReportModal';
-import { AlertObject } from './types';
+import { AlertObject, NotificationMethodOption } from './types';

 jest.mock('@superset-ui/core', () => ({
  ...jest.requireActual('@superset-ui/core'),
@ -30,7 +30,7 @@ jest.mock('@superset-ui/core', () => ({

 jest.mock('src/features/databases/state.ts', () => ({
  useCommonConf: () => ({
-    ALERT_REPORTS_NOTIFICATION_METHODS: ['Email', 'Slack'],
+    ALERT_REPORTS_NOTIFICATION_METHODS: ['Email', 'Slack', 'SlackV2'],
  }),
 }));

@ -135,7 +135,7 @@ const validAlert: AlertObject = {
  ],
  recipients: [
    {
-      type: 'Email',
+      type: NotificationMethodOption.Email,
      recipient_config_json: { target: 'test@user.com' },
    },
  ],
--- a/superset-frontend/src/features/alerts/AlertReportModal.tsx
+++ b/superset-frontend/src/features/alerts/AlertReportModal.tsx
@ -99,7 +99,9 @@ const DEFAULT_WORKING_TIMEOUT = 3600;
 const DEFAULT_CRON_VALUE = '0 0 * * *'; // every day
 const DEFAULT_RETENTION = 90;

-const DEFAULT_NOTIFICATION_METHODS: NotificationMethodOption[] = ['Email'];
+const DEFAULT_NOTIFICATION_METHODS: NotificationMethodOption[] = [
+  NotificationMethodOption.Email,
+];
 const DEFAULT_NOTIFICATION_FORMAT = 'PNG';
 const CONDITIONS = [
  {
@ -517,7 +519,7 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
    ]);

    setNotificationAddState(
-      notificationSettings.length === allowedNotificationMethods.length
+      notificationSettings.length === allowedNotificationMethodsCount
        ? 'hidden'
        : 'disabled',
    );
@ -1131,7 +1133,7 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
        {
          recipients: '',
          options: allowedNotificationMethods,
-          method: 'Email',
+          method: NotificationMethodOption.Email,
        },
      ]);
      setNotificationAddState('active');
@ -1235,6 +1237,20 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
    enforceValidation();
  }, [validationStatus]);

+  const allowedNotificationMethodsCount = useMemo(
+    () =>
+      allowedNotificationMethods.reduce((accum: string[], setting: string) => {
+        if (
+          accum.some(nm => nm.includes('slack')) &&
+          setting.toLowerCase().includes('slack')
+        ) {
+          return accum;
+        }
+        return [...accum, setting.toLowerCase()];
+      }, []).length,
+    [allowedNotificationMethods],
+  );
+
  // Show/hide
  if (isHidden && show) {
    setIsHidden(false);
@ -1743,7 +1759,7 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
          ))}
          {
            // Prohibit 'add notification method' button if only one present
-            allowedNotificationMethods.length > notificationSettings.length && (
+            allowedNotificationMethodsCount > notificationSettings.length && (
              <NotificationMethodAdd
                data-test="notification-add"
                status={notificationAddState}
--- a/superset-frontend/src/features/alerts/components/NotificationMethod.test.tsx
+++ b/superset-frontend/src/features/alerts/components/NotificationMethod.test.tsx
@ -0,0 +1,183 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { fireEvent, render, screen } from 'spec/helpers/testing-library';
+import userEvent from '@testing-library/user-event';
+
+import { NotificationMethod, mapSlackValues } from './NotificationMethod';
+import { NotificationMethodOption, NotificationSetting } from '../types';
+
+const mockOnUpdate = jest.fn();
+const mockOnRemove = jest.fn();
+const mockOnInputChange = jest.fn();
+const mockSetErrorSubject = jest.fn();
+
+const mockSetting: NotificationSetting = {
+  method: NotificationMethodOption.Email,
+  recipients: 'test@example.com',
+  options: [
+    NotificationMethodOption.Email,
+    NotificationMethodOption.Slack,
+    NotificationMethodOption.SlackV2,
+  ],
+};
+
+const mockEmailSubject = 'Test Subject';
+const mockDefaultSubject = 'Default Subject';
+
+describe('NotificationMethod', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('should render the component', () => {
+    render(
+      <NotificationMethod
+        setting={mockSetting}
+        index={0}
+        onUpdate={mockOnUpdate}
+        onRemove={mockOnRemove}
+        onInputChange={mockOnInputChange}
+        email_subject={mockEmailSubject}
+        defaultSubject={mockDefaultSubject}
+        setErrorSubject={mockSetErrorSubject}
+      />,
+    );
+
+    expect(screen.getByText('Notification Method')).toBeInTheDocument();
+    expect(
+      screen.getByText('Email subject name (optional)'),
+    ).toBeInTheDocument();
+    expect(screen.getByText('Email recipients')).toBeInTheDocument();
+  });
+
+  it('should call onRemove when the delete button is clicked', () => {
+    render(
+      <NotificationMethod
+        setting={mockSetting}
+        index={1}
+        onUpdate={mockOnUpdate}
+        onRemove={mockOnRemove}
+        onInputChange={mockOnInputChange}
+        email_subject={mockEmailSubject}
+        defaultSubject={mockDefaultSubject}
+        setErrorSubject={mockSetErrorSubject}
+      />,
+    );
+
+    const deleteButton = screen.getByRole('button');
+    userEvent.click(deleteButton);
+
+    expect(mockOnRemove).toHaveBeenCalledWith(1);
+  });
+
+  it('should update recipient value when input changes', () => {
+    render(
+      <NotificationMethod
+        setting={mockSetting}
+        index={0}
+        onUpdate={mockOnUpdate}
+        onRemove={mockOnRemove}
+        onInputChange={mockOnInputChange}
+        email_subject={mockEmailSubject}
+        defaultSubject={mockDefaultSubject}
+        setErrorSubject={mockSetErrorSubject}
+      />,
+    );
+
+    const recipientsInput = screen.getByTestId('recipients');
+    fireEvent.change(recipientsInput, {
+      target: { value: 'test1@example.com' },
+    });
+
+    expect(mockOnUpdate).toHaveBeenCalledWith(0, {
+      ...mockSetting,
+      recipients: 'test1@example.com',
+    });
+  });
+
+  it('should call onRecipientsChange when the recipients value is changed', () => {
+    render(
+      <NotificationMethod
+        setting={mockSetting}
+        index={0}
+        onUpdate={mockOnUpdate}
+        onRemove={mockOnRemove}
+        onInputChange={mockOnInputChange}
+        email_subject={mockEmailSubject}
+        defaultSubject={mockDefaultSubject}
+        setErrorSubject={mockSetErrorSubject}
+      />,
+    );
+
+    const recipientsInput = screen.getByTestId('recipients');
+    fireEvent.change(recipientsInput, {
+      target: { value: 'test1@example.com' },
+    });
+
+    expect(mockOnUpdate).toHaveBeenCalledWith(0, {
+      ...mockSetting,
+      recipients: 'test1@example.com',
+    });
+  });
+
+  it('should correctly map recipients when method is SlackV2', () => {
+    const method = 'SlackV2';
+    const recipientValue = 'user1,user2';
+    const slackOptions: { label: string; value: string }[] = [
+      { label: 'User One', value: 'user1' },
+      { label: 'User Two', value: 'user2' },
+    ];
+
+    const result = mapSlackValues({ method, recipientValue, slackOptions });
+
+    expect(result).toEqual([
+      { label: 'User One', value: 'user1' },
+      { label: 'User Two', value: 'user2' },
+    ]);
+  });
+
+  it('should return an empty array when recipientValue is an empty string', () => {
+    const method = 'SlackV2';
+    const recipientValue = '';
+    const slackOptions: { label: string; value: string }[] = [
+      { label: 'User One', value: 'user1' },
+      { label: 'User Two', value: 'user2' },
+    ];
+
+    const result = mapSlackValues({ method, recipientValue, slackOptions });
+
+    expect(result).toEqual([]);
+  });
+
+  it('should correctly map recipients when method is Slack with updated recipient values', () => {
+    const method = 'Slack';
+    const recipientValue = 'User One,User Two';
+    const slackOptions: { label: string; value: string }[] = [
+      { label: 'User One', value: 'user1' },
+      { label: 'User Two', value: 'user2' },
+    ];
+
+    const result = mapSlackValues({ method, recipientValue, slackOptions });
+
+    expect(result).toEqual([
+      { label: 'User One', value: 'user1' },
+      { label: 'User Two', value: 'user2' },
+    ]);
+  });
+});
--- a/superset-frontend/src/features/alerts/components/NotificationMethod.tsx
+++ b/superset-frontend/src/features/alerts/components/NotificationMethod.tsx
@ -16,12 +16,31 @@
 * specific language governing permissions and limitations
 * under the License.
 */
-import { FunctionComponent, useState, ChangeEvent } from 'react';
+import {
+  FunctionComponent,
+  useState,
+  ChangeEvent,
+  useEffect,
+  useMemo,
+} from 'react';
+import rison from 'rison';

-import { styled, t, useTheme } from '@superset-ui/core';
+import {
+  FeatureFlag,
+  JsonResponse,
+  SupersetClient,
+  isFeatureEnabled,
+  styled,
+  t,
+  useTheme,
+} from '@superset-ui/core';
 import { Select } from 'src/components';
 import Icons from 'src/components/Icons';
-import { NotificationMethodOption, NotificationSetting } from '../types';
+import {
+  NotificationMethodOption,
+  NotificationSetting,
+  SlackChannel,
+} from '../types';
 import { StyledInputContainer } from '../AlertReportModal';

 const StyledNotificationMethod = styled.div`
@ -73,6 +92,68 @@ const TRANSLATIONS = {
  ),
 };

+export const mapSlackValues = ({
+  method,
+  recipientValue,
+  slackOptions,
+}: {
+  method: string;
+  recipientValue: string;
+  slackOptions: { label: string; value: string }[];
+}) => {
+  const prop = method === NotificationMethodOption.SlackV2 ? 'value' : 'label';
+  return recipientValue
+    .split(',')
+    .map(recipient =>
+      slackOptions.find(
+        option =>
+          option[prop].trim().toLowerCase() === recipient.trim().toLowerCase(),
+      ),
+    )
+    .filter(val => !!val) as { label: string; value: string }[];
+};
+
+export const mapChannelsToOptions = (result: SlackChannel[]) => {
+  const publicChannels: SlackChannel[] = [];
+  const privateChannels: SlackChannel[] = [];
+
+  result.forEach(channel => {
+    if (channel.is_private) {
+      privateChannels.push(channel);
+    } else {
+      publicChannels.push(channel);
+    }
+  });
+
+  return [
+    {
+      label: 'Public Channels',
+      options: publicChannels.map((channel: SlackChannel) => ({
+        label: `${channel.name} ${
+          channel.is_member ? '' : t('(Bot not in channel)')
+        }`,
+        value: channel.id,
+        key: channel.id,
+      })),
+      key: 'public',
+    },
+    {
+      label: t('Private Channels (Bot in channel)'),
+      options: privateChannels.map((channel: SlackChannel) => ({
+        label: channel.name,
+        value: channel.id,
+        key: channel.id,
+      })),
+      key: 'private',
+    },
+  ];
+};
+
+type SlackOptionsType = {
+  label: string;
+  options: { label: string; value: string }[];
+}[];
+
 export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
  setting = null,
  index,
@ -87,20 +168,30 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
  const [recipientValue, setRecipientValue] = useState<string>(
    recipients || '',
  );
+  const [slackRecipients, setSlackRecipients] = useState<
+    { label: string; value: string }[]
+  >([]);
  const [error, setError] = useState(false);
  const theme = useTheme();
+  const [slackOptions, setSlackOptions] = useState<SlackOptionsType>([
+    {
+      label: '',
+      options: [],
+    },
+  ]);

-  if (!setting) {
-    return null;
-  }
+  const [useSlackV1, setUseSlackV1] = useState<boolean>(false);

-  const onMethodChange = (method: NotificationMethodOption) => {
+  const onMethodChange = (selected: {
+    label: string;
+    value: NotificationMethodOption;
+  }) => {
    // Since we're swapping the method, reset the recipients
    setRecipientValue('');
-    if (onUpdate) {
+    if (onUpdate && setting) {
      const updatedSetting = {
        ...setting,
-        method,
+        method: selected.value,
        recipients: '',
      };

@ -108,6 +199,94 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
    }
  };

+  const fetchSlackChannels = async ({
+    searchString = '',
+    types = [],
+    exactMatch = false,
+  }: {
+    searchString?: string | undefined;
+    types?: string[];
+    exactMatch?: boolean | undefined;
+  } = {}): Promise<JsonResponse> => {
+    const queryString = rison.encode({ searchString, types, exactMatch });
+    const endpoint = `/api/v1/report/slack_channels/?q=${queryString}`;
+    return SupersetClient.get({ endpoint });
+  };
+
+  useEffect(() => {
+    if (
+      method &&
+      [
+        NotificationMethodOption.Slack,
+        NotificationMethodOption.SlackV2,
+      ].includes(method) &&
+      !slackOptions[0]?.options.length
+    ) {
+      fetchSlackChannels({ types: ['public_channel', 'private_channel'] })
+        .then(({ json }) => {
+          const { result } = json;
+
+          const options: SlackOptionsType = mapChannelsToOptions(result);
+
+          setSlackOptions(options);
+
+          if (isFeatureEnabled(FeatureFlag.AlertReportSlackV2)) {
+            // map existing ids to names for display
+            // or names to ids if slack v1
+            const [publicOptions, privateOptions] = options;
+
+            setSlackRecipients(
+              mapSlackValues({
+                method,
+                recipientValue,
+                slackOptions: [
+                  ...publicOptions.options,
+                  ...privateOptions.options,
+                ],
+              }),
+            );
+            if (method === NotificationMethodOption.Slack) {
+              onMethodChange({
+                label: NotificationMethodOption.Slack,
+                value: NotificationMethodOption.SlackV2,
+              });
+            }
+          }
+        })
+        .catch(() => {
+          // Fallback to slack v1 if slack v2 is not compatible
+          setUseSlackV1(true);
+        });
+    }
+  }, [method]);
+
+  const methodOptions = useMemo(
+    () =>
+      (options || [])
+        .filter(
+          method =>
+            (isFeatureEnabled(FeatureFlag.AlertReportSlackV2) &&
+              !useSlackV1 &&
+              method === NotificationMethodOption.SlackV2) ||
+            ((!isFeatureEnabled(FeatureFlag.AlertReportSlackV2) ||
+              useSlackV1) &&
+              method === NotificationMethodOption.Slack) ||
+            method === NotificationMethodOption.Email,
+        )
+        .map(method => ({
+          label:
+            method === NotificationMethodOption.SlackV2
+              ? NotificationMethodOption.Slack
+              : method,
+          value: method,
+        })),
+    [options],
+  );
+
+  if (!setting) {
+    return null;
+  }
+
  const onRecipientsChange = (event: ChangeEvent<HTMLTextAreaElement>) => {
    const { target } = event;

@ -123,6 +302,21 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
    }
  };

+  const onSlackRecipientsChange = (
+    recipients: { label: string; value: string }[],
+  ) => {
+    setSlackRecipients(recipients);
+
+    if (onUpdate) {
+      const updatedSetting = {
+        ...setting,
+        recipients: recipients?.map(obj => obj.value).join(','),
+      };
+
+      onUpdate(index, updatedSetting);
+    }
+  };
+
  const onSubjectChange = (
    event: ChangeEvent<HTMLTextAreaElement | HTMLInputElement>,
  ) => {
@ -153,15 +347,12 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
            <Select
              ariaLabel={t('Delivery method')}
              data-test="select-delivery-method"
+              labelInValue
              onChange={onMethodChange}
              placeholder={t('Select Delivery Method')}
-              options={(options || []).map(
-                (method: NotificationMethodOption) => ({
-                  label: method,
-                  value: method,
-                }),
-              )}
-              value={method}
+              options={methodOptions}
+              showSearch
+              value={methodOptions.find(option => option.value === method)}
            />
            {index !== 0 && !!onRemove ? (
              <span
@ -180,7 +371,7 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
        <>
          <div className="inline-container">
            <StyledInputContainer>
-              {method === 'Email' ? (
+              {method === NotificationMethodOption.Email ? (
                <>
                  <div className="control-label">
                    {TRANSLATIONS.EMAIL_SUBJECT_NAME}
@ -211,19 +402,47 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
          <div className="inline-container">
            <StyledInputContainer>
              <div className="control-label">
-                {t('%s recipients', method)}
+                {t(
+                  '%s recipients',
+                  method === NotificationMethodOption.SlackV2
+                    ? NotificationMethodOption.Slack
+                    : method,
+                )}
                <span className="required">*</span>
              </div>
-              <div className="input-container">
-                <textarea
-                  name="recipients"
-                  data-test="recipients"
-                  value={recipientValue}
-                  onChange={onRecipientsChange}
-                />
-              </div>
-              <div className="helper">
-                {t('Recipients are separated by "," or ";"')}
+              <div>
+                {[
+                  NotificationMethodOption.Email,
+                  NotificationMethodOption.Slack,
+                ].includes(method) ? (
+                  <>
+                    <div className="input-container">
+                      <textarea
+                        name="recipients"
+                        data-test="recipients"
+                        value={recipientValue}
+                        onChange={onRecipientsChange}
+                      />
+                    </div>
+                    <div className="helper">
+                      {t('Recipients are separated by "," or ";"')}
+                    </div>
+                  </>
+                ) : (
+                  // for SlackV2
+                  <Select
+                    ariaLabel={t('Select channels')}
+                    mode="multiple"
+                    name="recipients"
+                    value={slackRecipients}
+                    options={slackOptions}
+                    onChange={onSlackRecipientsChange}
+                    allowClear
+                    data-test="recipients"
+                    allowSelectAll={false}
+                    labelInValue
+                  />
+                )}
              </div>
            </StyledInputContainer>
          </div>
--- a/superset-frontend/src/features/alerts/components/RecipientIcon.test.tsx
+++ b/superset-frontend/src/features/alerts/components/RecipientIcon.test.tsx
@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { render, screen } from 'spec/helpers/testing-library';
+import RecipientIcon from './RecipientIcon';
+import { NotificationMethodOption } from '../types';
+
+describe('RecipientIcon', () => {
+  it('should render the email icon when type is Email', () => {
+    render(<RecipientIcon type={NotificationMethodOption.Email} />);
+    const regexPattern = new RegExp(NotificationMethodOption.Email, 'i');
+    const emailIcon = screen.getByLabelText(regexPattern);
+    expect(emailIcon).toBeInTheDocument();
+  });
+
+  it('should render the Slack icon when type is Slack', () => {
+    render(<RecipientIcon type={NotificationMethodOption.Slack} />);
+    const regexPattern = new RegExp(NotificationMethodOption.Slack, 'i');
+    const slackIcon = screen.getByLabelText(regexPattern);
+    expect(slackIcon).toBeInTheDocument();
+  });
+
+  it('should render the Slack icon when type is SlackV2', () => {
+    render(<RecipientIcon type={NotificationMethodOption.SlackV2} />);
+    const regexPattern = new RegExp(NotificationMethodOption.Slack, 'i');
+    const slackIcon = screen.getByLabelText(regexPattern);
+    expect(slackIcon).toBeInTheDocument();
+  });
+
+  it('should not render any icon when type is not recognized', () => {
+    render(<RecipientIcon type="unknown" />);
+    const icons = screen.queryByLabelText(/.*/);
+    expect(icons).not.toBeInTheDocument();
+  });
+});
--- a/superset-frontend/src/features/alerts/components/RecipientIcon.tsx
+++ b/superset-frontend/src/features/alerts/components/RecipientIcon.tsx
@ -20,7 +20,7 @@ import { SupersetTheme, css } from '@superset-ui/core';
 import { ReactElement } from 'react';
 import { Tooltip } from 'src/components/Tooltip';
 import Icons from 'src/components/Icons';
-import { RecipientIconName } from '../types';
+import { NotificationMethodOption } from '../types';

 const StyledIcon = (theme: SupersetTheme) => css`
  color: ${theme.colors.grayscale.light1};
@ -33,13 +33,17 @@ export default function RecipientIcon({ type }: { type: string }) {
    label: '',
  };
  switch (type) {
-    case RecipientIconName.Email:
+    case NotificationMethodOption.Email:
      recipientIconConfig.icon = <Icons.Email css={StyledIcon} />;
-      recipientIconConfig.label = RecipientIconName.Email;
+      recipientIconConfig.label = NotificationMethodOption.Email;
      break;
-    case RecipientIconName.Slack:
+    case NotificationMethodOption.Slack:
      recipientIconConfig.icon = <Icons.Slack css={StyledIcon} />;
-      recipientIconConfig.label = RecipientIconName.Slack;
+      recipientIconConfig.label = NotificationMethodOption.Slack;
+      break;
+    case NotificationMethodOption.SlackV2:
+      recipientIconConfig.icon = <Icons.Slack css={StyledIcon} />;
+      recipientIconConfig.label = NotificationMethodOption.Slack;
      break;
    default:
      recipientIconConfig.icon = null;
--- a/superset-frontend/src/features/alerts/types.ts
+++ b/superset-frontend/src/features/alerts/types.ts
@ -41,7 +41,11 @@ export type DatabaseObject = {
  id: number;
 };

-export type NotificationMethodOption = 'Email' | 'Slack';
+export enum NotificationMethodOption {
+  Email = 'Email',
+  Slack = 'Slack',
+  SlackV2 = 'SlackV2',
+}

 export type NotificationSetting = {
  method?: NotificationMethodOption;
@ -49,6 +53,13 @@ export type NotificationSetting = {
  options: NotificationMethodOption[];
 };

+export type SlackChannel = {
+  id: string;
+  name: string;
+  is_member: boolean;
+  is_private: boolean;
+};
+
 export type Recipient = {
  recipient_config_json: {
    target: string;
@ -124,6 +135,7 @@ export enum AlertState {
 export enum RecipientIconName {
  Email = 'Email',
  Slack = 'Slack',
+  SlackV2 = 'SlackV2',
 }
 export interface AlertsReportsConfig {
  ALERT_REPORTS_DEFAULT_WORKING_TIMEOUT: number;
--- a/superset/commands/report/execute.py
+++ b/superset/commands/report/execute.py
@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
+from copy import deepcopy
 from datetime import datetime, timedelta
 from typing import Any, Optional, Union
 from uuid import UUID
@ -25,7 +26,7 @@ from celery.exceptions import SoftTimeLimitExceeded
 from superset import app, db, security_manager
 from superset.commands.base import BaseCommand
 from superset.commands.dashboard.permalink.create import CreateDashboardPermalinkCommand
-from superset.commands.exceptions import CommandException
+from superset.commands.exceptions import CommandException, UpdateFailedError
 from superset.commands.report.alert import AlertCommand
 from superset.commands.report.exceptions import (
    ReportScheduleAlertGracePeriodError,
@ -64,7 +65,10 @@ from superset.reports.models import (
 )
 from superset.reports.notifications import create_notification
 from superset.reports.notifications.base import NotificationContent
-from superset.reports.notifications.exceptions import NotificationError
+from superset.reports.notifications.exceptions import (
+    NotificationError,
+    SlackV1NotificationError,
+)
 from superset.tasks.utils import get_executor
 from superset.utils import json
 from superset.utils.core import HeaderDataType, override_user
@ -72,6 +76,7 @@ from superset.utils.csv import get_chart_csv_data, get_chart_dataframe
 from superset.utils.decorators import logs_context, transaction
 from superset.utils.pdf import build_pdf_from_screenshots
 from superset.utils.screenshots import ChartScreenshot, DashboardScreenshot
+from superset.utils.slack import get_channels_with_search, SlackChannelTypes
 from superset.utils.urls import get_url_path

 logger = logging.getLogger(__name__)
@ -121,6 +126,40 @@ class BaseReportState:
        self._report_schedule.last_state = state
        self._report_schedule.last_eval_dttm = datetime.utcnow()

+    def update_report_schedule_slack_v2(self) -> None:
+        """
+        Update the report schedule type and channels for all slack recipients to v2.
+        V2 uses ids instead of names for channels.
+        """
+        try:
+            updated_recipients = []
+            for recipient in self._report_schedule.recipients:
+                recipient_copy = deepcopy(recipient)
+                if recipient_copy.type == ReportRecipientType.SLACK:
+                    recipient_copy.type = ReportRecipientType.SLACKV2
+                    slack_recipients = json.loads(recipient_copy.recipient_config_json)
+                    # we need to ensure that existing reports can also fetch
+                    # ids from private channels
+                    recipient_copy.recipient_config_json = json.dumps(
+                        {
+                            "target": get_channels_with_search(
+                                slack_recipients["target"],
+                                types=[
+                                    SlackChannelTypes.PRIVATE,
+                                    SlackChannelTypes.PUBLIC,
+                                ],
+                            )
+                        }
+                    )
+
+                updated_recipients.append(recipient_copy)
+            db.session.commit()  # pylint: disable=consider-using-transaction
+        except Exception as ex:
+            logger.warning(
+                "Failed to update slack recipients to v2: %s", str(ex), exc_info=True
+            )
+            raise UpdateFailedError from ex
+
    def create_log(self, error_message: Optional[str] = None) -> None:
        """
        Creates a Report execution log, uses the current computed last_value for Alerts
@ -439,6 +478,19 @@ class BaseReportState:
                    )
                else:
                    notification.send()
+            except SlackV1NotificationError as ex:
+                # The slack notification should be sent with the v2 api
+                logger.info("Attempting to upgrade the report to Slackv2: %s", str(ex))
+                try:
+                    self.update_report_schedule_slack_v2()
+                    recipient.type = ReportRecipientType.SLACKV2
+                    notification = create_notification(recipient, notification_content)
+                    notification.send()
+                except UpdateFailedError as err:
+                    # log the error but keep processing the report with SlackV1
+                    logger.warning(
+                        "Failed to update slack recipients to v2: %s", str(err)
+                    )
            except (NotificationError, SupersetException) as ex:
                # collect errors but keep processing them
                notification_errors.append(
--- a/superset/config.py
+++ b/superset/config.py
@ -483,6 +483,7 @@ DEFAULT_FEATURE_FLAGS: dict[str, bool] = {
    # Enables Alerts and reports new implementation
    "ALERT_REPORTS": False,
    "ALERT_REPORT_TABS": False,
+    "ALERT_REPORT_SLACK_V2": False,
    "DASHBOARD_RBAC": False,
    "ENABLE_ADVANCED_DATA_TYPES": False,
    # Enabling ALERTS_ATTACH_REPORTS, the system sends email and slack message
--- a/superset/constants.py
+++ b/superset/constants.py
@ -172,6 +172,7 @@ MODEL_API_RW_METHOD_PERMISSION_MAP = {
    "excel_metadata": "excel_upload",
    "columnar_metadata": "columnar_upload",
    "csv_metadata": "csv_upload",
+    "slack_channels": "write",
 }

 EXTRA_FORM_DATA_APPEND_KEYS = {
--- a/superset/reports/api.py
+++ b/superset/reports/api.py
@ -40,15 +40,18 @@ from superset.commands.report.update import UpdateReportScheduleCommand
 from superset.constants import MODEL_API_RW_METHOD_PERMISSION_MAP, RouteMethod
 from superset.dashboards.filters import DashboardAccessFilter
 from superset.databases.filters import DatabaseFilter
+from superset.exceptions import SupersetException
 from superset.extensions import event_logger
 from superset.reports.filters import ReportScheduleAllTextFilter, ReportScheduleFilter
 from superset.reports.models import ReportSchedule
 from superset.reports.schemas import (
    get_delete_ids_schema,
+    get_slack_channels_schema,
    openapi_spec_methods_override,
    ReportSchedulePostSchema,
    ReportSchedulePutSchema,
 )
+from superset.utils.slack import get_channels_with_search
 from superset.views.base_api import (
    BaseSupersetModelRestApi,
    RelatedFieldFilter,
@ -71,7 +74,8 @@ class ReportScheduleRestApi(BaseSupersetModelRestApi):

    include_route_methods = RouteMethod.REST_MODEL_VIEW_CRUD_SET | {
        RouteMethod.RELATED,
-        "bulk_delete",  # not using RouteMethod since locally defined
+        "bulk_delete",
+        "slack_channels",  # not using RouteMethod since locally defined
    }
    class_permission_name = "ReportSchedule"
    method_permission_name = MODEL_API_RW_METHOD_PERMISSION_MAP
@ -513,3 +517,68 @@ class ReportScheduleRestApi(BaseSupersetModelRestApi):
            return self.response_403()
        except ReportScheduleDeleteFailedError as ex:
            return self.response_422(message=str(ex))
+
+    @expose("/slack_channels/", methods=("GET",))
+    @protect()
+    @rison(get_slack_channels_schema)
+    @safe
+    @statsd_metrics
+    @event_logger.log_this_with_context(
+        action=lambda self,
+        *args,
+        **kwargs: f"{self.__class__.__name__}.slack_channels",
+        log_to_statsd=False,
+    )
+    def slack_channels(self, **kwargs: Any) -> Response:
+        """Get slack channels.
+        ---
+        get:
+          summary: Get slack channels
+          description: Get slack channels
+          parameters:
+            - in: query
+              name: q
+              content:
+                application/json:
+                  schema:
+                    $ref: '#/components/schemas/get_slack_channels_schema'
+          responses:
+            200:
+              description: Slack channels
+              content:
+                application/json:
+                  schema:
+                    type: object
+                    properties:
+                      result:
+                        type: array
+                        items:
+                          type: object
+                          properties:
+                            id:
+                              type: string
+                            name:
+                              type: string
+            401:
+              $ref: '#/components/responses/401'
+            403:
+              $ref: '#/components/responses/403'
+            404:
+              $ref: '#/components/responses/404'
+            422:
+              $ref: '#/components/responses/422'
+            500:
+              $ref: '#/components/responses/500'
+        """
+        try:
+            params = kwargs.get("rison", {})
+            search_string = params.get("search_string")
+            types = params.get("types", [])
+            exact_match = params.get("exact_match", False)
+            channels = get_channels_with_search(
+                search_string=search_string, types=types, exact_match=exact_match
+            )
+            return self.response(200, result=channels)
+        except SupersetException as ex:
+            logger.error("Error fetching slack channels %s", str(ex))
+            return self.response_422(message=str(ex))
--- a/superset/reports/models.py
+++ b/superset/reports/models.py
@ -62,6 +62,7 @@ class ReportScheduleValidatorType(StrEnum):
 class ReportRecipientType(StrEnum):
    EMAIL = "Email"
    SLACK = "Slack"
+    SLACKV2 = "SlackV2"


 class ReportState(StrEnum):
--- a/superset/reports/notifications/init.py
+++ b/superset/reports/notifications/init.py
@ -18,6 +18,7 @@ from superset.reports.models import ReportRecipients
 from superset.reports.notifications.base import BaseNotification, NotificationContent
 from superset.reports.notifications.email import EmailNotification  # noqa: F401
 from superset.reports.notifications.slack import SlackNotification  # noqa: F401
+from superset.reports.notifications.slackv2 import SlackV2Notification  # noqa: F401


 def create_notification(
--- a/superset/reports/notifications/email.py
+++ b/superset/reports/notifications/email.py
@ -135,7 +135,7 @@ class EmailNotification(BaseNotification):  # pylint: disable=too-few-public-met
        for msgid in images.keys():
            img_tags.append(
                f"""<div class="image">
-                    <img width="1000px" src="cid:{msgid}">
+                    <img width="1000" src="cid:{msgid}">
                </div>
                """
            )
@ -153,6 +153,7 @@ class EmailNotification(BaseNotification):  # pylint: disable=too-few-public-met
                  }}
                  .image{{
                      margin-bottom: 18px;
+                      min-width: 1000px;
                  }}
                </style>
              </head>
--- a/superset/reports/notifications/exceptions.py
+++ b/superset/reports/notifications/exceptions.py
@ -24,6 +24,17 @@ class NotificationError(SupersetException):
    """


+class SlackV1NotificationError(SupersetException):
+    """
+    Report should not be run with the slack v1 api
+    """
+
+    message = """Report should not be run with the Slack V1 api.
+    Attempting to run with V2 if required Slack scopes are available"""
+
+    status = 422
+
+
 class NotificationParamException(SupersetException):
    status = 422

--- a/superset/reports/notifications/slack.py
+++ b/superset/reports/notifications/slack.py
@ -17,14 +17,10 @@
 import logging
 from collections.abc import Sequence
 from io import IOBase
-from typing import List, Union
+from typing import Union

 import backoff
-import pandas as pd
-from deprecation import deprecated
 from flask import g
-from flask_babel import gettext as __
-from slack_sdk import WebClient
 from slack_sdk.errors import (
    BotUserAccessError,
    SlackApiError,
@ -43,172 +39,68 @@ from superset.reports.notifications.exceptions import (
    NotificationMalformedException,
    NotificationParamException,
    NotificationUnprocessableException,
+    SlackV1NotificationError,
 )
+from superset.reports.notifications.slack_mixin import SlackMixin
 from superset.utils import json
 from superset.utils.core import get_email_address_list
 from superset.utils.decorators import statsd_gauge
-from superset.utils.slack import get_slack_client
+from superset.utils.slack import (
+    get_slack_client,
+    should_use_v2_api,
+)

 logger = logging.getLogger(__name__)

-# Slack only allows Markdown messages up to 4k chars
-MAXIMUM_MESSAGE_SIZE = 4000

-
-class SlackNotification(BaseNotification):  # pylint: disable=too-few-public-methods
+# TODO: Deprecated: Remove this class in Superset 6.0.0
+class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-few-public-methods
    """
    Sends a slack notification for a report recipient
    """

    type = ReportRecipientType.SLACK

-    def _get_channels(self, client: WebClient) -> List[str]:
+    def _get_channel(self) -> str:
        """
        Get the recipient's channel(s).
-        :returns: A list of channel ids: "EID676L"
-        :raises SlackApiError: If the API call fails
+        Note Slack SDK uses "channel" to refer to one or more
+        channels. Multiple channels are demarcated by a comma.
+        :returns: The comma separated list of channel(s)
        """
        recipient_str = json.loads(self._recipient.recipient_config_json)["target"]

-        channel_recipients: List[str] = get_email_address_list(recipient_str)
-
-        conversations_list_response = client.conversations_list(
-            types="public_channel,private_channel"
-        )
-
-        return [
-            c["id"]
-            for c in conversations_list_response["channels"]
-            if c["name"] in channel_recipients
-        ]
-
-    def _message_template(self, table: str = "") -> str:
-        return __(
-            """*%(name)s*
-
-%(description)s
-
-<%(url)s|Explore in Superset>
-
-%(table)s
-""",
-            name=self._content.name,
-            description=self._content.description or "",
-            url=self._content.url,
-            table=table,
-        )
-
-    @staticmethod
-    def _error_template(name: str, description: str, text: str) -> str:
-        return __(
-            """*%(name)s*
-
-%(description)s
-
-Error: %(text)s
-""",
-            name=name,
-            description=description,
-            text=text,
-        )
-
-    def _get_body(self) -> str:
-        if self._content.text:
-            return self._error_template(
-                self._content.name, self._content.description or "", self._content.text
-            )
-
-        if self._content.embedded_data is None:
-            return self._message_template()
-
-        # Embed data in the message
-        df = self._content.embedded_data
-
-        # Flatten columns/index so they show up nicely in the table
-        df.columns = [
-            (
-                " ".join(str(name) for name in column).strip()
-                if isinstance(column, tuple)
-                else column
-            )
-            for column in df.columns
-        ]
-        df.index = [
-            (
-                " ".join(str(name) for name in index).strip()
-                if isinstance(index, tuple)
-                else index
-            )
-            for index in df.index
-        ]
-
-        # Slack Markdown only works on messages shorter than 4k chars, so we might
-        # need to truncate the data
-        for i in range(len(df) - 1):
-            truncated_df = df[: i + 1].fillna("")
-            truncated_row = pd.Series({k: "..." for k in df.columns})
-            truncated_df = pd.concat(
-                [truncated_df, truncated_row.to_frame().T], ignore_index=True
-            )
-            tabulated = df.to_markdown()
-            table = f"```\n{tabulated}\n```\n\n(table was truncated)"
-            message = self._message_template(table)
-            if len(message) > MAXIMUM_MESSAGE_SIZE:
-                # Decrement i and build a message that is under the limit
-                truncated_df = df[:i].fillna("")
-                truncated_row = pd.Series({k: "..." for k in df.columns})
-                truncated_df = pd.concat(
-                    [truncated_df, truncated_row.to_frame().T], ignore_index=True
-                )
-                tabulated = df.to_markdown()
-                table = (
-                    f"```\n{tabulated}\n```\n\n(table was truncated)"
-                    if len(truncated_df) > 0
-                    else ""
-                )
-                break
-
-        # Send full data
-        else:
-            tabulated = df.to_markdown()
-            table = f"```\n{tabulated}\n```"
-
-        return self._message_template(table)
+        return ",".join(get_email_address_list(recipient_str))

    def _get_inline_files(
        self,
-    ) -> Sequence[Union[str, IOBase, bytes]]:
+    ) -> tuple[Union[str, None], Sequence[Union[str, IOBase, bytes]]]:
        if self._content.csv:
-            return [self._content.csv]
+            return ("csv", [self._content.csv])
        if self._content.screenshots:
-            return self._content.screenshots
+            return ("png", self._content.screenshots)
        if self._content.pdf:
-            return [self._content.pdf]
-        return []
+            return ("pdf", [self._content.pdf])
+        return (None, [])

-    @deprecated(deprecated_in="4.1")
-    def _deprecated_upload_files(
-        self, client: WebClient, title: str, body: str
-    ) -> None:
-        """
-        Deprecated method to upload files to slack
-        Should only be used if the new method fails
-        To be removed in the next major release
-        """
-        file_type, files = (None, [])
-        if self._content.csv:
-            file_type, files = ("csv", [self._content.csv])
-        if self._content.screenshots:
-            file_type, files = ("png", self._content.screenshots)
-        if self._content.pdf:
-            file_type, files = ("pdf", [self._content.pdf])
+    @backoff.on_exception(backoff.expo, SlackApiError, factor=10, base=2, max_tries=5)
+    @statsd_gauge("reports.slack.send")
+    def send(self) -> None:
+        file_type, files = self._get_inline_files()
+        title = self._content.name
+        body = self._get_body(content=self._content)
+        global_logs_context = getattr(g, "logs_context", {}) or {}

-        recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+        # see if the v2 api will work
+        if should_use_v2_api():
+            # if we can fetch channels, then raise an error and use the v2 api
+            raise SlackV1NotificationError

-        recipients = get_email_address_list(recipient_str)
-
-        for channel in recipients:
-            if len(files) > 0:
+        try:
+            client = get_slack_client()
+            channel = self._get_channel()
+            # files_upload returns SlackResponse as we run it in sync mode.
+            if files:
                for file in files:
                    client.files_upload(
                        channels=channel,
@ -219,46 +111,6 @@ Error: %(text)s
                    )
            else:
                client.chat_postMessage(channel=channel, text=body)
-
-    @backoff.on_exception(backoff.expo, SlackApiError, factor=10, base=2, max_tries=5)
-    @statsd_gauge("reports.slack.send")
-    def send(self) -> None:
-        global_logs_context = getattr(g, "logs_context", {}) or {}
-        try:
-            client = get_slack_client()
-            title = self._content.name
-            body = self._get_body()
-
-            try:
-                channels = self._get_channels(client)
-            except SlackApiError:
-                logger.warning(
-                    "Slack scope missing. Using deprecated API to get channels. Please update your Slack app to use the new API.",
-                    extra={
-                        "execution_id": global_logs_context.get("execution_id"),
-                    },
-                )
-                self._deprecated_upload_files(client, title, body)
-                return
-
-            if channels == []:
-                raise NotificationParamException("No valid channel found")
-
-            files = self._get_inline_files()
-
-            # files_upload returns SlackResponse as we run it in sync mode.
-            for channel in channels:
-                if len(files) > 0:
-                    for file in files:
-                        client.files_upload_v2(
-                            channel=channel,
-                            file=file,
-                            initial_comment=body,
-                            title=title,
-                        )
-                else:
-                    client.chat_postMessage(channel=channel, text=body)
-
            logger.info(
                "Report sent to slack",
                extra={
--- a/superset/reports/notifications/slack_mixin.py
+++ b/superset/reports/notifications/slack_mixin.py
@ -0,0 +1,124 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pandas as pd
+from flask_babel import gettext as __
+
+from superset.reports.notifications.base import NotificationContent
+
+# Slack only allows Markdown messages up to 4k chars
+MAXIMUM_MESSAGE_SIZE = 4000
+
+
+# pylint: disable=too-few-public-methods
+class SlackMixin:
+    def _message_template(
+        self,
+        content: NotificationContent,
+        table: str = "",
+    ) -> str:
+        return __(
+            """*%(name)s*
+
+%(description)s
+
+<%(url)s|Explore in Superset>
+
+%(table)s
+""",
+            name=content.name,
+            description=content.description or "",
+            url=content.url,
+            table=table,
+        )
+
+    @staticmethod
+    def _error_template(name: str, description: str, text: str) -> str:
+        return __(
+            """*%(name)s*
+
+    %(description)s
+
+    Error: %(text)s
+    """,
+            name=name,
+            description=description,
+            text=text,
+        )
+
+    def _get_body(self, content: NotificationContent) -> str:
+        if content.text:
+            return self._error_template(
+                content.name, content.description or "", content.text
+            )
+
+        if content.embedded_data is None:
+            return self._message_template(content=content)
+
+        # Embed data in the message
+        df = content.embedded_data
+
+        # Flatten columns/index so they show up nicely in the table
+        df.columns = [
+            (
+                " ".join(str(name) for name in column).strip()
+                if isinstance(column, tuple)
+                else column
+            )
+            for column in df.columns
+        ]
+        df.index = [
+            (
+                " ".join(str(name) for name in index).strip()
+                if isinstance(index, tuple)
+                else index
+            )
+            for index in df.index
+        ]
+
+        # Slack Markdown only works on messages shorter than 4k chars, so we might
+        # need to truncate the data
+        for i in range(len(df) - 1):
+            truncated_df = df[: i + 1].fillna("")
+            truncated_row = pd.Series({k: "..." for k in df.columns})
+            truncated_df = pd.concat(
+                [truncated_df, truncated_row.to_frame().T], ignore_index=True
+            )
+            tabulated = df.to_markdown()
+            table = f"```\n{tabulated}\n```\n\n(table was truncated)"
+            message = self._message_template(table=table, content=content)
+            if len(message) > MAXIMUM_MESSAGE_SIZE:
+                # Decrement i and build a message that is under the limit
+                truncated_df = df[:i].fillna("")
+                truncated_row = pd.Series({k: "..." for k in df.columns})
+                truncated_df = pd.concat(
+                    [truncated_df, truncated_row.to_frame().T], ignore_index=True
+                )
+                tabulated = df.to_markdown()
+                table = (
+                    f"```\n{tabulated}\n```\n\n(table was truncated)"
+                    if len(truncated_df) > 0
+                    else ""
+                )
+                break
+
+        # Send full data
+        else:
+            tabulated = df.to_markdown()
+            table = f"```\n{tabulated}\n```"
+
+        return self._message_template(table=table, content=content)
--- a/superset/reports/notifications/slackv2.py
+++ b/superset/reports/notifications/slackv2.py
@ -0,0 +1,130 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+import logging
+from collections.abc import Sequence
+from io import IOBase
+from typing import List, Union
+
+import backoff
+from flask import g
+from slack_sdk.errors import (
+    BotUserAccessError,
+    SlackApiError,
+    SlackClientConfigurationError,
+    SlackClientError,
+    SlackClientNotConnectedError,
+    SlackObjectFormationError,
+    SlackRequestError,
+    SlackTokenRotationError,
+)
+
+from superset.reports.models import ReportRecipientType
+from superset.reports.notifications.base import BaseNotification
+from superset.reports.notifications.exceptions import (
+    NotificationAuthorizationException,
+    NotificationMalformedException,
+    NotificationParamException,
+    NotificationUnprocessableException,
+)
+from superset.reports.notifications.slack_mixin import SlackMixin
+from superset.utils import json
+from superset.utils.core import get_email_address_list
+from superset.utils.decorators import statsd_gauge
+from superset.utils.slack import get_slack_client
+
+logger = logging.getLogger(__name__)
+
+
+class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-few-public-methods
+    """
+    Sends a slack notification for a report recipient with the slack upload v2 API
+    """
+
+    type = ReportRecipientType.SLACKV2
+
+    def _get_channels(self) -> List[str]:
+        """
+        Get the recipient's channel(s).
+        :returns: A list of channel ids: "EID676L"
+        :raises NotificationParamException or SlackApiError: If the recipient is not found
+        """
+        recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+
+        return get_email_address_list(recipient_str)
+
+    def _get_inline_files(
+        self,
+    ) -> Sequence[Union[str, IOBase, bytes]]:
+        if self._content.csv:
+            return [self._content.csv]
+        if self._content.screenshots:
+            return self._content.screenshots
+        if self._content.pdf:
+            return [self._content.pdf]
+        return []
+
+    @backoff.on_exception(backoff.expo, SlackApiError, factor=10, base=2, max_tries=5)
+    @statsd_gauge("reports.slack.send")
+    def send(self) -> None:
+        global_logs_context = getattr(g, "logs_context", {}) or {}
+        try:
+            client = get_slack_client()
+            title = self._content.name
+            body = self._get_body(content=self._content)
+
+            channels = self._get_channels()
+
+            if not channels:
+                raise NotificationParamException("No recipients saved in the report")
+
+            files = self._get_inline_files()
+
+            # files_upload returns SlackResponse as we run it in sync mode.
+            for channel in channels:
+                if len(files) > 0:
+                    for file in files:
+                        client.files_upload_v2(
+                            channel=channel,
+                            file=file,
+                            initial_comment=body,
+                            title=title,
+                        )
+                else:
+                    client.chat_postMessage(channel=channel, text=body)
+
+            logger.info(
+                "Report sent to slack",
+                extra={
+                    "execution_id": global_logs_context.get("execution_id"),
+                },
+            )
+        except (
+            BotUserAccessError,
+            SlackRequestError,
+            SlackClientConfigurationError,
+        ) as ex:
+            raise NotificationParamException(str(ex)) from ex
+        except SlackObjectFormationError as ex:
+            raise NotificationMalformedException(str(ex)) from ex
+        except SlackTokenRotationError as ex:
+            raise NotificationAuthorizationException(str(ex)) from ex
+        except (SlackClientNotConnectedError, SlackApiError) as ex:
+            raise NotificationUnprocessableException(str(ex)) from ex
+        except SlackClientError as ex:
+            # this is the base class for all slack client errors
+            # keep it last so that it doesn't interfere with @backoff
+            raise NotificationUnprocessableException(str(ex)) from ex
--- a/superset/reports/schemas.py
+++ b/superset/reports/schemas.py
@ -49,6 +49,17 @@ openapi_spec_methods_override = {
 }

 get_delete_ids_schema = {"type": "array", "items": {"type": "integer"}}
+get_slack_channels_schema = {
+    "type": "object",
+    "properties": {
+        "search_string": {"type": "string"},
+        "types": {
+            "type": "array",
+            "items": {"type": "string", "enum": ["public_channel", "private_channel"]},
+        },
+        "exact_match": {"type": "boolean"},
+    },
+}

 type_description = "The report schedule type"
 name_description = "The report schedule name."
--- a/superset/tasks/scheduler.py
+++ b/superset/tasks/scheduler.py
@ -94,7 +94,7 @@ def execute(self: Celery.task, report_schedule_id: int) -> None:
        ).run()
    except ReportScheduleUnexpectedError:
        logger.exception(
-            "An unexpected occurred while executing the report: %s", task_id
+            "An unexpected error occurred while executing the report: %s", task_id
        )
        self.update_state(state="FAILURE")
    except CommandException as ex:
--- a/superset/utils/slack.py
+++ b/superset/utils/slack.py
@ -16,8 +16,23 @@
 # under the License.


+import logging
+from typing import Optional
+
 from flask import current_app
 from slack_sdk import WebClient
+from slack_sdk.errors import SlackApiError
+
+from superset import feature_flag_manager
+from superset.exceptions import SupersetException
+from superset.utils.backports import StrEnum
+
+logger = logging.getLogger(__name__)
+
+
+class SlackChannelTypes(StrEnum):
+    PUBLIC = "public_channel"
+    PRIVATE = "private_channel"


 class SlackClientError(Exception):
@ -31,6 +46,80 @@ def get_slack_client() -> WebClient:
    return WebClient(token=token, proxy=current_app.config["SLACK_PROXY"])


+def get_channels_with_search(
+    search_string: str = "",
+    limit: int = 999,
+    types: Optional[list[SlackChannelTypes]] = None,
+    exact_match: bool = False,
+) -> list[str]:
+    """
+    The slack api is paginated but does not include search, so we need to fetch
+    all channels and filter them ourselves
+    This will search by slack name or id
+    """
+
+    try:
+        client = get_slack_client()
+        channels = []
+        cursor = None
+        extra_params = {}
+        extra_params["types"] = ",".join(types) if types else None
+
+        while True:
+            response = client.conversations_list(
+                limit=limit, cursor=cursor, exclude_archived=True, **extra_params
+            )
+            channels.extend(response.data["channels"])
+            cursor = response.data.get("response_metadata", {}).get("next_cursor")
+            if not cursor:
+                break
+
+        # The search string can be multiple channels separated by commas
+        if search_string:
+            search_array = [
+                search.lower()
+                for search in (search_string.split(",") if search_string else [])
+            ]
+
+            channels = [
+                channel
+                for channel in channels
+                if any(
+                    (
+                        search == channel["name"].lower()
+                        or search == channel["id"].lower()
+                        if exact_match
+                        else (
+                            search in channel["name"].lower()
+                            or search in channel["id"].lower()
+                        )
+                    )
+                    for search in search_array
+                )
+            ]
+        return channels
+    except (SlackClientError, SlackApiError) as ex:
+        raise SupersetException(f"Failed to list channels: {ex}") from ex
+
+
+def should_use_v2_api() -> bool:
+    if not feature_flag_manager.is_feature_enabled("ALERT_REPORT_SLACK_V2"):
+        return False
+    try:
+        client = get_slack_client()
+        client.conversations_list()
+        logger.info("Slack API v2 is available")
+        return True
+    except SlackApiError:
+        # use the v1 api but warn with a deprecation message
+        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."""
+        )
+        return False
+
+
 def get_user_avatar(email: str, client: WebClient = None) -> str:
    client = client or get_slack_client()
    try:
--- a/superset/views/base.py
+++ b/superset/views/base.py
@ -316,6 +316,7 @@ def cached_common_bootstrap_data(  # pylint: disable=unused-argument
        frontend_config["ALERT_REPORTS_NOTIFICATION_METHODS"] = [
            ReportRecipientType.EMAIL,
            ReportRecipientType.SLACK,
+            ReportRecipientType.SLACKV2,
        ]
    else:
        frontend_config["ALERT_REPORTS_NOTIFICATION_METHODS"] = [
--- a/tests/integration_tests/fixtures/world_bank_dashboard.py
+++ b/tests/integration_tests/fixtures/world_bank_dashboard.py
@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 import string
+from operator import or_
 from random import choice, randint, random, uniform
 from typing import Any

@ -146,16 +147,14 @@ def _cleanup(dash_id: int, slices_ids: list[int]) -> None:


 def _cleanup_reports(dash_id: int, slices_ids: list[int]) -> None:
-    reports_with_dash = (
-        db.session.query(ReportSchedule).filter_by(dashboard_id=dash_id).all()
-    )
-    reports_with_slices = (
-        db.session.query(ReportSchedule)
-        .filter(ReportSchedule.chart_id.in_(slices_ids))
-        .all()
+    reports = db.session.query(ReportSchedule).filter(
+        or_(
+            ReportSchedule.dashboard_id == dash_id,
+            ReportSchedule.chart_id.in_(slices_ids),
+        )
    )

-    for report in reports_with_dash + reports_with_slices:
+    for report in reports:
        db.session.delete(report)
    db.session.commit()

--- a/tests/integration_tests/reports/commands_tests.py
+++ b/tests/integration_tests/reports/commands_tests.py
@ -268,6 +268,17 @@ def create_report_slack_chart():
    cleanup_report_schedule(report_schedule)


+@pytest.fixture()
+def create_report_slack_chartv2():
+    chart = db.session.query(Slice).first()
+    report_schedule = create_report_notification(
+        slack_channel="slack_channel_id", chart=chart, name="report_slack_chartv2"
+    )
+    yield report_schedule
+
+    cleanup_report_schedule(report_schedule)
+
+
@pytest.fixture()
 def create_report_slack_chart_with_csv():
    chart = db.session.query(Slice).first()
@ -1100,13 +1111,17 @@ def test_email_dashboard_report_schedule_force_screenshot(


@pytest.mark.usefixtures(
-    "load_birth_names_dashboard_with_slices", "create_report_slack_chart"
+    "load_birth_names_dashboard_with_slices", "create_report_slack_chartv2"
 )
-@patch("superset.reports.notifications.slack.get_slack_client")
+@patch("superset.commands.report.execute.get_channels_with_search")
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
+@patch("superset.reports.notifications.slackv2.get_slack_client")
@patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
-def test_slack_chart_report_schedule(
+def test_slack_chart_report_schedule_v2(
    screenshot_mock,
    slack_client_mock,
+    slack_should_use_v2_api_mock,
+    get_channels_with_search_mock,
    create_report_slack_chart,
 ):
    """
@ -1116,11 +1131,9 @@ def test_slack_chart_report_schedule(
    screenshot_mock.return_value = SCREENSHOT_FILE
    notification_targets = get_target_from_report_schedule(create_report_slack_chart)

-    channel_name = notification_targets[0]
-    channel_id = "channel_id"
-    slack_client_mock.return_value.conversations_list.return_value = {
-        "channels": [{"id": channel_id, "name": channel_name}]
-    }
+    channel_id = notification_targets[0]
+
+    get_channels_with_search_mock.return_value = {}

    with freeze_time("2020-01-01T00:00:00Z"):
        with patch.object(current_app.config["STATS_LOGGER"], "gauge") as statsd_mock:
@ -1139,56 +1152,17 @@ def test_slack_chart_report_schedule(

            # Assert logs are correct
            assert_log(ReportState.SUCCESS)
-            statsd_mock.assert_called_once_with("reports.slack.send.ok", 1)
+            # this will send a warning
+            assert statsd_mock.call_args_list[0] == call(
+                "reports.slack.send.warning", 1
+            )
+            assert statsd_mock.call_args_list[1] == call("reports.slack.send.ok", 1)


@pytest.mark.usefixtures(
    "load_birth_names_dashboard_with_slices", "create_report_slack_chart"
 )
-@patch("superset.reports.notifications.slack.get_slack_client")
-@patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
-def test_slack_chart_report_schedule_deprecated(
-    screenshot_mock,
-    slack_client_mock,
-    create_report_slack_chart,
-):
-    """
-    ExecuteReport Command: Test chart slack report schedule
-    """
-    # setup screenshot mock
-    screenshot_mock.return_value = SCREENSHOT_FILE
-    notification_targets = get_target_from_report_schedule(create_report_slack_chart)
-
-    channel_name = notification_targets[0]
-
-    slack_client_mock.return_value.conversations_list.side_effect = SlackApiError(
-        "Error", "Response"
-    )
-
-    with freeze_time("2020-01-01T00:00:00Z"):
-        with patch.object(current_app.config["STATS_LOGGER"], "gauge") as statsd_mock:
-            AsyncExecuteReportScheduleCommand(
-                TEST_ID, create_report_slack_chart.id, datetime.utcnow()
-            ).run()
-
-            assert (
-                slack_client_mock.return_value.files_upload.call_args[1]["channels"]
-                == channel_name
-            )
-            assert (
-                slack_client_mock.return_value.files_upload.call_args[1]["file"]
-                == SCREENSHOT_FILE
-            )
-
-            # Assert logs are correct
-            assert_log(ReportState.SUCCESS)
-            statsd_mock.assert_called_once_with("reports.slack.send.ok", 1)
-
-
-@pytest.mark.usefixtures(
-    "load_birth_names_dashboard_with_slices", "create_report_slack_chart"
-)
-@patch("superset.utils.slack.WebClient")
+@patch("superset.utils.slack.get_slack_client")
@patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
 def test_slack_chart_report_schedule_with_errors(
    screenshot_mock,
@ -1214,7 +1188,7 @@ def test_slack_chart_report_schedule_with_errors(
    ]

    for idx, er in enumerate(slack_errors):
-        web_client_mock.side_effect = er
+        web_client_mock.side_effect = [SlackApiError(None, None), er]

        with pytest.raises(ReportScheduleClientErrorsException):
            AsyncExecuteReportScheduleCommand(
@ -1242,6 +1216,7 @@ def test_slack_chart_report_schedule_with_errors(
@pytest.mark.usefixtures(
    "load_birth_names_dashboard_with_slices", "create_report_slack_chart_with_csv"
 )
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=False)
@patch("superset.reports.notifications.slack.get_slack_client")
@patch("superset.utils.csv.urllib.request.urlopen")
@patch("superset.utils.csv.urllib.request.OpenerDirector.open")
@ -1251,10 +1226,11 @@ def test_slack_chart_report_schedule_with_csv(
    mock_open,
    mock_urlopen,
    slack_client_mock_class,
+    slack_should_use_v2_api_mock,
    create_report_slack_chart_with_csv,
 ):
    """
-    ExecuteReport Command: Test chart slack report schedule with CSV
+    ExecuteReport Command: Test chart slack report V1 schedule with CSV
    """
    # setup csv mock
    response = Mock()
@ -1268,63 +1244,6 @@ def test_slack_chart_report_schedule_with_csv(
    )

    channel_name = notification_targets[0]
-    channel_id = "channel_id"
-    slack_client_mock_class.return_value = Mock()
-    slack_client_mock_class.return_value.conversations_list.return_value = {
-        "channels": [{"id": channel_id, "name": channel_name}]
-    }
-
-    with freeze_time("2020-01-01T00:00:00Z"):
-        AsyncExecuteReportScheduleCommand(
-            TEST_ID, create_report_slack_chart_with_csv.id, datetime.utcnow()
-        ).run()
-
-        assert (
-            slack_client_mock_class.return_value.files_upload_v2.call_args[1]["channel"]
-            == channel_id
-        )
-        assert (
-            slack_client_mock_class.return_value.files_upload_v2.call_args[1]["file"]
-            == CSV_FILE
-        )
-
-        # Assert logs are correct
-        assert_log(ReportState.SUCCESS)
-
-
-@pytest.mark.usefixtures(
-    "load_birth_names_dashboard_with_slices", "create_report_slack_chart_with_csv"
-)
-@patch("superset.reports.notifications.slack.get_slack_client")
-@patch("superset.utils.csv.urllib.request.urlopen")
-@patch("superset.utils.csv.urllib.request.OpenerDirector.open")
-@patch("superset.utils.csv.get_chart_csv_data")
-def test_slack_chart_report_schedule_with_csv_deprecated_api(
-    csv_mock,
-    mock_open,
-    mock_urlopen,
-    slack_client_mock_class,
-    create_report_slack_chart_with_csv,
-):
-    """
-    ExecuteReport Command: Test chart slack report schedule with CSV
-    """
-    # setup csv mock
-    response = Mock()
-    mock_open.return_value = response
-    mock_urlopen.return_value = response
-    mock_urlopen.return_value.getcode.return_value = 200
-    response.read.return_value = CSV_FILE
-
-    notification_targets = get_target_from_report_schedule(
-        create_report_slack_chart_with_csv
-    )
-
-    channel_name = notification_targets[0]
-    slack_client_mock_class.return_value = Mock()
-    slack_client_mock_class.return_value.conversations_list.side_effect = SlackApiError(
-        "Error", "Response"
-    )

    with freeze_time("2020-01-01T00:00:00Z"):
        AsyncExecuteReportScheduleCommand(
@ -1347,6 +1266,7 @@ def test_slack_chart_report_schedule_with_csv_deprecated_api(
@pytest.mark.usefixtures(
    "load_birth_names_dashboard_with_slices", "create_report_slack_chart_with_text"
 )
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=False)
@patch("superset.utils.csv.urllib.request.urlopen")
@patch("superset.utils.csv.urllib.request.OpenerDirector.open")
@patch("superset.reports.notifications.slack.get_slack_client")
@ -1356,6 +1276,7 @@ def test_slack_chart_report_schedule_with_text(
    slack_client_mock_class,
    mock_open,
    mock_urlopen,
+    slack_should_use_v2_api_mock,
    create_report_slack_chart_with_text,
 ):
    """
@ -1383,17 +1304,6 @@ def test_slack_chart_report_schedule_with_text(
        }
    ).encode("utf-8")

-    notification_targets = get_target_from_report_schedule(
-        create_report_slack_chart_with_text
-    )
-
-    channel_name = notification_targets[0]
-    channel_id = "channel_id"
-
-    slack_client_mock_class.return_value.conversations_list.return_value = {
-        "channels": [{"id": channel_id, "name": channel_name}]
-    }
-
    with freeze_time("2020-01-01T00:00:00Z"):
        AsyncExecuteReportScheduleCommand(
            TEST_ID, create_report_slack_chart_with_text.id, datetime.utcnow()
@ -1420,87 +1330,6 @@ def test_slack_chart_report_schedule_with_text(
        assert_log(ReportState.SUCCESS)


-@pytest.mark.usefixtures(
-    "load_birth_names_dashboard_with_slices", "create_report_slack_chart_with_text"
-)
-@patch("superset.utils.csv.urllib.request.urlopen")
-@patch("superset.utils.csv.urllib.request.OpenerDirector.open")
-@patch("superset.reports.notifications.slack.get_slack_client")
-@patch("superset.utils.csv.get_chart_dataframe")
-def test_slack_chart_report_schedule_with_text_deprecated_slack_api(
-    dataframe_mock,
-    slack_client_mock_class,
-    mock_open,
-    mock_urlopen,
-    create_report_slack_chart_with_text,
-):
-    """
-    ExecuteReport Command: Test chart slack report schedule with text
-    """
-    # setup dataframe mock
-    response = Mock()
-    mock_open.return_value = response
-    mock_urlopen.return_value = response
-    mock_urlopen.return_value.getcode.return_value = 200
-    response.read.return_value = json.dumps(
-        {
-            "result": [
-                {
-                    "data": {
-                        "t1": {0: "c11", 1: "c21"},
-                        "t2": {0: "c12", 1: "c22"},
-                        "t3__sum": {0: "c13", 1: "c23"},
-                    },
-                    "colnames": [("t1",), ("t2",), ("t3__sum",)],
-                    "indexnames": [(0,), (1,)],
-                    "coltypes": [1, 1, 0],
-                },
-            ],
-        }
-    ).encode("utf-8")
-
-    notification_targets = get_target_from_report_schedule(
-        create_report_slack_chart_with_text
-    )
-
-    channel_name = notification_targets[0]
-
-    slack_client_mock_class.return_value.conversations_list.side_effect = SlackApiError(
-        "Error", "Response"
-    )
-
-    with freeze_time("2020-01-01T00:00:00Z"):
-        AsyncExecuteReportScheduleCommand(
-            TEST_ID, create_report_slack_chart_with_text.id, datetime.utcnow()
-        ).run()
-
-        table_markdown = """|    | t1   | t2   | t3__sum   |
-|---:|:-----|:-----|:----------|
-|  0 | c11  | c12  | c13       |
-|  1 | c21  | c22  | c23       |"""
-        assert (
-            table_markdown
-            in slack_client_mock_class.return_value.chat_postMessage.call_args[1][
-                "text"
-            ]
-        )
-        assert (
-            f"<http://0.0.0.0:8080/explore/?form_data=%7B%22slice_id%22:+{create_report_slack_chart_with_text.chart.id}%7D&force=false|Explore in Superset>"
-            in slack_client_mock_class.return_value.chat_postMessage.call_args[1][
-                "text"
-            ]
-        )
-        assert (
-            slack_client_mock_class.return_value.chat_postMessage.call_args[1][
-                "channel"
-            ]
-            == channel_name
-        )
-
-        # Assert logs are correct
-        assert_log(ReportState.SUCCESS)
-
-
@pytest.mark.usefixtures("create_report_slack_chart")
 def test_report_schedule_not_found(create_report_slack_chart):
    """
--- a/tests/unit_tests/notifications/init.py
+++ b/tests/unit_tests/notifications/init.py
@ -1,16 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#   http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.
--- a/tests/unit_tests/notifications/slack_tests.py
+++ b/tests/unit_tests/notifications/slack_tests.py
@ -1,88 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#   http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.
-import uuid
-from unittest.mock import MagicMock, patch
-
-import pandas as pd
-
-
-@patch("superset.reports.notifications.slack.g")
-@patch("superset.reports.notifications.slack.logger")
-@patch("superset.reports.notifications.slack.get_slack_client")
-def test_send_slack(
-    slack_client_mock: MagicMock,
-    logger_mock: MagicMock,
-    flask_global_mock: MagicMock,
-) -> None:
-    # `superset.models.helpers`, a dependency of following imports,
-    # requires app context
-    from superset.reports.models import ReportRecipients, ReportRecipientType
-    from superset.reports.notifications.base import NotificationContent
-    from superset.reports.notifications.slack import SlackNotification
-
-    execution_id = uuid.uuid4()
-    flask_global_mock.logs_context = {"execution_id": execution_id}
-    slack_client_mock.return_value.conversations_list.return_value = {
-        "channels": [{"name": "some_channel", "id": "123"}]
-    }
-    content = NotificationContent(
-        name="test alert",
-        header_data={
-            "notification_format": "PNG",
-            "notification_type": "Alert",
-            "owners": [1],
-            "notification_source": None,
-            "chart_id": None,
-            "dashboard_id": None,
-        },
-        embedded_data=pd.DataFrame(
-            {
-                "A": [1, 2, 3],
-                "B": [4, 5, 6],
-                "C": ["111", "222", '<a href="http://www.example.com">333</a>'],
-            }
-        ),
-        description='<p>This is <a href="#">a test</a> alert</p><br />',
-    )
-
-    SlackNotification(
-        recipient=ReportRecipients(
-            type=ReportRecipientType.SLACK,
-            recipient_config_json='{"target": "some_channel"}',
-        ),
-        content=content,
-    ).send()
-    logger_mock.info.assert_called_with(
-        "Report sent to slack", extra={"execution_id": execution_id}
-    )
-    slack_client_mock.return_value.chat_postMessage.assert_called_with(
-        channel="123",
-        text="""*test alert*
-
-<p>This is <a href="#">a test</a> alert</p><br />
-
-<None|Explore in Superset>
-
-```
-|    |   A |   B | C                                        |
-|---:|----:|----:|:-----------------------------------------|
-|  0 |   1 |   4 | 111                                      |
-|  1 |   2 |   5 | 222                                      |
-|  2 |   3 |   6 | <a href="http://www.example.com">333</a> |
-```
-""",
-    )
--- a/tests/unit_tests/reports/notifications/email_tests.py
+++ b/tests/unit_tests/reports/notifications/email_tests.py
--- a/tests/unit_tests/reports/notifications/slack_tests.py
+++ b/tests/unit_tests/reports/notifications/slack_tests.py
@ -14,9 +14,14 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from unittest.mock import Mock
+
+import uuid
+from unittest.mock import MagicMock, patch

 import pandas as pd
+from slack_sdk.errors import SlackApiError
+
+from superset.reports.notifications.slackv2 import SlackV2Notification


 def test_get_channel_with_multi_recipients() -> None:
@ -55,15 +60,350 @@ def test_get_channel_with_multi_recipients() -> None:
        content=content,
    )

-    client = Mock()
-    client.conversations_list.return_value = {
-        "channels": [
-            {"name": "some_channel", "id": "23SDKE"},
-            {"name": "second_channel", "id": "WD3D8KE"},
-            {"name": "third_channel", "id": "223DFKE"},
-        ]
+    result = slack_notification._get_channel()
+
+    assert result == "some_channel,second_channel,third_channel"
+
+    # Test if the recipient configuration JSON is valid when using a SlackV2 recipient type
+
+
+def test_valid_recipient_config_json_slackv2() -> None:
+    """
+    Test if the recipient configuration JSON is valid when using a SlackV2 recipient type
+    """
+    from superset.reports.models import ReportRecipients, ReportRecipientType
+    from superset.reports.notifications.base import NotificationContent
+    from superset.reports.notifications.slack import SlackNotification
+
+    content = NotificationContent(
+        name="test alert",
+        header_data={
+            "notification_format": "PNG",
+            "notification_type": "Alert",
+            "owners": [1],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+        },
+        embedded_data=pd.DataFrame(
+            {
+                "A": [1, 2, 3],
+                "B": [4, 5, 6],
+                "C": ["111", "222", '<a href="http://www.example.com">333</a>'],
+            }
+        ),
+        description='<p>This is <a href="#">a test</a> alert</p><br />',
+    )
+    slack_notification = SlackNotification(
+        recipient=ReportRecipients(
+            type=ReportRecipientType.SLACKV2,
+            recipient_config_json='{"target": "some_channel"}',
+        ),
+        content=content,
+    )
+
+    result = slack_notification._recipient.recipient_config_json
+
+    assert result == '{"target": "some_channel"}'
+
+    # Ensure _get_inline_files function returns the correct tuple when content has screenshots
+
+
+def test_get_inline_files_with_screenshots() -> None:
+    """
+    Test the _get_inline_files function to ensure it will return the correct tuple
+    when content has screenshots
+    """
+    from superset.reports.models import ReportRecipients, ReportRecipientType
+    from superset.reports.notifications.base import NotificationContent
+    from superset.reports.notifications.slack import SlackNotification
+
+    content = NotificationContent(
+        name="test alert",
+        header_data={
+            "notification_format": "PNG",
+            "notification_type": "Alert",
+            "owners": [1],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+        },
+        embedded_data=pd.DataFrame(
+            {
+                "A": [1, 2, 3],
+                "B": [4, 5, 6],
+                "C": ["111", "222", '<a href="http://www.example.com">333</a>'],
+            }
+        ),
+        description='<p>This is <a href="#">a test</a> alert</p><br />',
+        screenshots=[b"screenshot1", b"screenshot2"],
+    )
+    slack_notification = SlackNotification(
+        recipient=ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json='{"target": "some_channel"}',
+        ),
+        content=content,
+    )
+
+    result = slack_notification._get_inline_files()
+
+    assert result == ("png", [b"screenshot1", b"screenshot2"])
+
+    # Ensure _get_inline_files function returns None when content has no screenshots or csv
+
+
+def test_get_inline_files_with_no_screenshots_or_csv() -> None:
+    """
+    Test the _get_inline_files function to ensure it will return None
+    when content has no screenshots or csv
+    """
+    from superset.reports.models import ReportRecipients, ReportRecipientType
+    from superset.reports.notifications.base import NotificationContent
+    from superset.reports.notifications.slack import SlackNotification
+
+    content = NotificationContent(
+        name="test alert",
+        header_data={
+            "notification_format": "PNG",
+            "notification_type": "Alert",
+            "owners": [1],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+        },
+        embedded_data=pd.DataFrame(
+            {
+                "A": [1, 2, 3],
+                "B": [4, 5, 6],
+                "C": ["111", "222", '<a href="http://www.example.com">333</a>'],
+            }
+        ),
+        description='<p>This is <a href="#">a test</a> alert</p><br />',
+    )
+    slack_notification = SlackNotification(
+        recipient=ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json='{"target": "some_channel"}',
+        ),
+        content=content,
+    )
+
+    result = slack_notification._get_inline_files()
+
+    assert result == (None, [])
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.logger")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_send_slackv2(
+    slack_client_mock: MagicMock,
+    logger_mock: MagicMock,
+    flask_global_mock: MagicMock,
+) -> None:
+    # `superset.models.helpers`, a dependency of following imports,
+    # requires app context
+    from superset.reports.models import ReportRecipients, ReportRecipientType
+    from superset.reports.notifications.base import NotificationContent
+
+    execution_id = uuid.uuid4()
+    flask_global_mock.logs_context = {"execution_id": execution_id}
+    slack_client_mock.return_value.chat_postMessage.return_value = {"ok": True}
+    content = NotificationContent(
+        name="test alert",
+        header_data={
+            "notification_format": "PNG",
+            "notification_type": "Alert",
+            "owners": [1],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+        },
+        embedded_data=pd.DataFrame(
+            {
+                "A": [1, 2, 3],
+                "B": [4, 5, 6],
+                "C": ["111", "222", '<a href="http://www.example.com">333</a>'],
+            }
+        ),
+        description='<p>This is <a href="#">a test</a> alert</p><br />',
+    )
+
+    notification = SlackV2Notification(
+        recipient=ReportRecipients(
+            type=ReportRecipientType.SLACKV2,
+            recipient_config_json='{"target": "some_channel"}',
+        ),
+        content=content,
+    )
+    notification.send()
+    logger_mock.info.assert_called_with(
+        "Report sent to slack", extra={"execution_id": execution_id}
+    )
+    slack_client_mock.return_value.chat_postMessage.assert_called_with(
+        channel="some_channel",
+        text="""*test alert*
+
+<p>This is <a href="#">a test</a> alert</p><br />
+
+<None|Explore in Superset>
+
+```
+|    |   A |   B | C                                        |
+|---:|----:|----:|:-----------------------------------------|
+|  0 |   1 |   4 | 111                                      |
+|  1 |   2 |   5 | 222                                      |
+|  2 |   3 |   6 | <a href="http://www.example.com">333</a> |
+```
+""",
+    )
+
+
+@patch("superset.reports.notifications.slack.g")
+@patch("superset.reports.notifications.slack.logger")
+@patch("superset.utils.slack.get_slack_client")
+@patch("superset.reports.notifications.slack.get_slack_client")
+def test_send_slack(
+    slack_client_mock: MagicMock,
+    slack_client_mock_util: MagicMock,
+    logger_mock: MagicMock,
+    flask_global_mock: MagicMock,
+) -> None:
+    # `superset.models.helpers`, a dependency of following imports,
+    # requires app context
+    from superset.reports.models import ReportRecipients, ReportRecipientType
+    from superset.reports.notifications.base import NotificationContent
+    from superset.reports.notifications.slack import SlackNotification
+
+    execution_id = uuid.uuid4()
+    flask_global_mock.logs_context = {"execution_id": execution_id}
+    slack_client_mock.return_value.chat_postMessage.return_value = {"ok": True}
+    slack_client_mock_util.return_value.conversations_list.side_effect = SlackApiError(
+        "scope not found", "error"
+    )
+
+    content = NotificationContent(
+        name="test alert",
+        header_data={
+            "notification_format": "PNG",
+            "notification_type": "Alert",
+            "owners": [1],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+        },
+        embedded_data=pd.DataFrame(
+            {
+                "A": [1, 2, 3],
+                "B": [4, 5, 6],
+                "C": ["111", "222", '<a href="http://www.example.com">333</a>'],
+            }
+        ),
+        description='<p>This is <a href="#">a test</a> alert</p><br />',
+    )
+
+    notification = SlackNotification(
+        recipient=ReportRecipients(
+            type=ReportRecipientType.SLACKV2,
+            recipient_config_json='{"target": "some_channel"}',
+        ),
+        content=content,
+    )
+    notification.send()
+
+    logger_mock.info.assert_called_with(
+        "Report sent to slack", extra={"execution_id": execution_id}
+    )
+    slack_client_mock.return_value.chat_postMessage.assert_called_with(
+        channel="some_channel",
+        text="""*test alert*
+
+<p>This is <a href="#">a test</a> alert</p><br />
+
+<None|Explore in Superset>
+
+```
+|    |   A |   B | C                                        |
+|---:|----:|----:|:-----------------------------------------|
+|  0 |   1 |   4 | 111                                      |
+|  1 |   2 |   5 | 222                                      |
+|  2 |   3 |   6 | <a href="http://www.example.com">333</a> |
+```
+""",
+    )
+
+
+@patch("superset.reports.notifications.slack.g")
+@patch("superset.reports.notifications.slack.logger")
+@patch("superset.utils.slack.get_slack_client")
+@patch("superset.reports.notifications.slack.get_slack_client")
+def test_send_slack_no_feature_flag(
+    slack_client_mock: MagicMock,
+    slack_client_mock_util: MagicMock,
+    logger_mock: MagicMock,
+    flask_global_mock: MagicMock,
+) -> None:
+    # `superset.models.helpers`, a dependency of following imports,
+    # requires app context
+    from superset.reports.models import ReportRecipients, ReportRecipientType
+    from superset.reports.notifications.base import NotificationContent
+    from superset.reports.notifications.slack import SlackNotification
+
+    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
+    slack_client_mock_util.return_value.conversations_list.return_value = {
+        "channels": [{"id": "foo", "name": "bar"}]
    }

-    result = slack_notification._get_channels(client)
+    content = NotificationContent(
+        name="test alert",
+        header_data={
+            "notification_format": "PNG",
+            "notification_type": "Alert",
+            "owners": [1],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+        },
+        embedded_data=pd.DataFrame(
+            {
+                "A": [1, 2, 3],
+                "B": [4, 5, 6],
+                "C": ["111", "222", '<a href="http://www.example.com">333</a>'],
+            }
+        ),
+        description='<p>This is <a href="#">a test</a> alert</p><br />',
+    )

-    assert result == ["23SDKE", "WD3D8KE", "223DFKE"]
+    notification = SlackNotification(
+        recipient=ReportRecipients(
+            type=ReportRecipientType.SLACKV2,
+            recipient_config_json='{"target": "some_channel"}',
+        ),
+        content=content,
+    )
+    notification.send()
+
+    logger_mock.info.assert_called_with(
+        "Report sent to slack", extra={"execution_id": execution_id}
+    )
+    slack_client_mock.return_value.chat_postMessage.assert_called_with(
+        channel="some_channel",
+        text="""*test alert*
+
+<p>This is <a href="#">a test</a> alert</p><br />
+
+<None|Explore in Superset>
+
+```
+|    |   A |   B | C                                        |
+|---:|----:|----:|:-----------------------------------------|
+|  0 |   1 |   4 | 111                                      |
+|  1 |   2 |   5 | 222                                      |
+|  2 |   3 |   6 | <a href="http://www.example.com">333</a> |
+```
+""",
+    )
--- a/tests/unit_tests/utils/slack_test.py
+++ b/tests/unit_tests/utils/slack_test.py
@ -0,0 +1,193 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pytest
+
+from superset.utils.slack import get_channels_with_search
+
+
+class MockResponse:
+    def __init__(self, data):
+        self._data = data
+
+    @property
+    def data(self):
+        return self._data
+
+
+class TestGetChannelsWithSearch:
+    # Fetch all channels when no search string is provided
+    def test_fetch_all_channels_no_search_string(self, mocker):
+        # Mock data
+        mock_data = {
+            "channels": [{"name": "general", "id": "C12345"}],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        # Mock class instance with data property
+        mock_response_instance = MockResponse(mock_data)
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response_instance
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search()
+        assert result == [{"name": "general", "id": "C12345"}]
+
+    # Handle an empty search string gracefully
+    def test_handle_empty_search_string(self, mocker):
+        mock_data = {
+            "channels": [{"name": "general", "id": "C12345"}],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        mock_response_instance = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response_instance
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(search_string="")
+        assert result == [{"name": "general", "id": "C12345"}]
+
+    def test_handle_exact_match_search_string_single_channel(self, mocker):
+        # Mock data with multiple channels
+        mock_data = {
+            "channels": [
+                {"name": "general", "id": "C12345"},
+                {"name": "general2", "id": "C13454"},
+                {"name": "random", "id": "C67890"},
+            ],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        # Mock response and client setup
+        mock_response_instance = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response_instance
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        # Call the function with a search string that matches a single channel
+        result = get_channels_with_search(search_string="general", exact_match=True)
+
+        # Assert that the result is a list with a single channel dictionary
+        assert result == [{"name": "general", "id": "C12345"}]
+
+    def test_handle_exact_match_search_string_multiple_channels(self, mocker):
+        mock_data = {
+            "channels": [
+                {"name": "general", "id": "C12345"},
+                {"name": "general2", "id": "C13454"},
+                {"name": "random", "id": "C67890"},
+            ],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        mock_response_instance = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response_instance
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(
+            search_string="general,random", exact_match=True
+        )
+        assert result == [
+            {"name": "general", "id": "C12345"},
+            {"name": "random", "id": "C67890"},
+        ]
+
+    def test_handle_loose_match_search_string_multiple_channels(self, mocker):
+        mock_data = {
+            "channels": [
+                {"name": "general", "id": "C12345"},
+                {"name": "general2", "id": "C13454"},
+                {"name": "random", "id": "C67890"},
+            ],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        mock_response_instance = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response_instance
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(search_string="general,random")
+        assert result == [
+            {"name": "general", "id": "C12345"},
+            {"name": "general2", "id": "C13454"},
+            {"name": "random", "id": "C67890"},
+        ]
+
+    def test_handle_slack_client_error_listing_channels(self, mocker):
+        from slack_sdk.errors import SlackApiError
+
+        from superset.exceptions import SupersetException
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = SlackApiError(
+            "foo", "missing scope: channels:read"
+        )
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        with pytest.raises(SupersetException) as ex:
+            get_channels_with_search()
+
+        assert str(ex.value) == (
+            """Failed to list channels: foo
+The server responded with: missing scope: channels:read"""
+        )
+
+    def test_filter_channels_by_specified_types(self, mocker):
+        mock_data = {
+            "channels": [
+                {"name": "general", "id": "C12345", "type": "public"},
+            ],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        mock_response_instance = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response_instance
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(types=["public"])
+        assert result == [{"name": "general", "id": "C12345", "type": "public"}]
+
+    def test_handle_pagination_multiple_pages(self, mocker):
+        mock_data_page1 = {
+            "channels": [{"name": "general", "id": "C12345"}],
+            "response_metadata": {"next_cursor": "page2"},
+        }
+        mock_data_page2 = {
+            "channels": [{"name": "random", "id": "C67890"}],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        mock_response_instance_page1 = MockResponse(mock_data_page1)
+        mock_response_instance_page2 = MockResponse(mock_data_page2)
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = [
+            mock_response_instance_page1,
+            mock_response_instance_page2,
+        ]
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search()
+        assert result == [
+            {"name": "general", "id": "C12345"},
+            {"name": "random", "id": "C67890"},
+        ]